diff --git a/src/common/backend/port/unix_latch.cpp b/src/common/backend/port/unix_latch.cpp index 2a02a0daf..44fd96d5c 100644 --- a/src/common/backend/port/unix_latch.cpp +++ b/src/common/backend/port/unix_latch.cpp @@ -526,8 +526,9 @@ void ResetLatch(volatile Latch* latch) */ void latch_sigusr1_handler(void) { - if (waiting) + if (waiting) { sendSelfPipeByte(); + } } /* Send one byte to the self-pipe, to wake up WaitLatch */ diff --git a/src/common/backend/utils/init/globals.cpp b/src/common/backend/utils/init/globals.cpp index e0818b9a7..e0601f566 100755 --- a/src/common/backend/utils/init/globals.cpp +++ b/src/common/backend/utils/init/globals.cpp @@ -51,6 +51,7 @@ THR_LOCAL object_access_hook_type object_access_hook = NULL; * These are initialized for the bootstrap/standalone case. */ THR_LOCAL bool IsUnderPostmaster = false; +THR_LOCAL bool IsBackgroundWorker = false; volatile ThreadId PostmasterPid = 0; bool IsPostmasterEnvironment = false; diff --git a/src/common/backend/utils/init/miscinit.cpp b/src/common/backend/utils/init/miscinit.cpp index 1aab5afed..8eb284208 100755 --- a/src/common/backend/utils/init/miscinit.cpp +++ b/src/common/backend/utils/init/miscinit.cpp @@ -724,11 +724,11 @@ bool has_rolvcadmin(Oid role_id) /* * Initialize user identity during normal backend startup */ -void InitializeSessionUserId(const char* role_name) +void InitializeSessionUserId(const char* role_name, Oid role_id) { HeapTuple role_tup; Form_pg_authid rform; - Oid role_id; + char* rname = NULL; /* Audit user login */ char details[PGAUDIT_MAXLENGTH]; @@ -744,23 +744,33 @@ void InitializeSessionUserId(const char* role_name) AssertState(!OidIsValid(u_sess->misc_cxt.AuthenticatedUserId)); } - role_tup = SearchSysCache1(AUTHNAME, PointerGetDatum(role_name)); - if (!HeapTupleIsValid(role_tup)) { - /* Audit user login */ - int rcs = snprintf_truncated_s(details, - sizeof(details), - "login db(%s) failed-the role(%s)does not exist", - u_sess->proc_cxt.MyProcPort->database_name, - role_name); - securec_check_ss(rcs, "", ""); - pgaudit_user_login(FALSE, u_sess->proc_cxt.MyProcPort->database_name, details); + if (role_name != NULL) { + role_tup = SearchSysCache1(AUTHNAME, PointerGetDatum(role_name)); + if (!HeapTupleIsValid(role_tup)) { + /* Audit user login */ + int rcs = snprintf_truncated_s(details, + sizeof(details), + "login db(%s) failed-the role(%s)does not exist", + u_sess->proc_cxt.MyProcPort->database_name, + role_name); + securec_check_ss(rcs, "", ""); + pgaudit_user_login(FALSE, u_sess->proc_cxt.MyProcPort->database_name, details); - ereport(FATAL, (errcode(ERRCODE_INVALID_AUTHORIZATION_SPECIFICATION), - errmsg("Invalid username/password,login denied."))); + ereport(FATAL, (errcode(ERRCODE_INVALID_AUTHORIZATION_SPECIFICATION), + errmsg("Invalid username/password,login denied."))); + } + } else { + role_tup = SearchSysCache1(AUTHOID, ObjectIdGetDatum(role_id)); + if (!HeapTupleIsValid(role_tup)) { + ereport(FATAL, + (errcode(ERRCODE_INVALID_AUTHORIZATION_SPECIFICATION), + errmsg("role with OID %u does not exist", role_id))); + } } rform = (Form_pg_authid)GETSTRUCT(role_tup); role_id = HeapTupleGetOid(role_tup); + rname = NameStr(rform->rolname); u_sess->misc_cxt.AuthenticatedUserId = role_id; u_sess->misc_cxt.AuthenticatedUserIsSuperuser = rform->rolsuper; @@ -832,10 +842,11 @@ void InitializeSessionUserIdStandalone(void) { /* * This function should only be called in single-user mode and in - * autovacuum workers. + * autovacuum workers, and in background workers. */ AssertState(!IsUnderPostmaster || IsAutoVacuumWorkerProcess() || - IsJobSchedulerProcess() || IsJobWorkerProcess() || AM_WAL_SENDER); + IsJobSchedulerProcess() || IsJobWorkerProcess() || AM_WAL_SENDER || + IsBackgroundWorker); /* In pooler stateless reuse mode, to reset session userid */ if (!g_instance.attr.attr_network.PoolerStatelessReuse) { diff --git a/src/common/backend/utils/init/postinit.cpp b/src/common/backend/utils/init/postinit.cpp index b8b4a5fd5..2775810b9 100644 --- a/src/common/backend/utils/init/postinit.cpp +++ b/src/common/backend/utils/init/postinit.cpp @@ -683,7 +683,7 @@ void PostgresResetUsernamePgoption(const char* username) u_sess->proc_cxt.MyProcPort->user_name = (char*)GetSuperUserName((char*)username); } - InitializeSessionUserId(username); + InitializeSessionUserId(username, InvalidOid); am_superuser = superuser(); u_sess->misc_cxt.CurrentUserName = u_sess->proc_cxt.MyProcPort->user_name; } @@ -1059,6 +1059,7 @@ PostgresInitializer::PostgresInitializer() m_indbname = NULL; m_dboid = InvalidOid; m_username = NULL; + m_useroid = InvalidOid; m_isSuperUser = false; m_fullpath = NULL; memset_s(m_dbname, NAMEDATALEN, 0, NAMEDATALEN); @@ -1074,11 +1075,13 @@ PostgresInitializer::~PostgresInitializer() m_username = NULL; } -void PostgresInitializer::SetDatabaseAndUser(const char* in_dbname, Oid dboid, const char* username) +void PostgresInitializer::SetDatabaseAndUser( + const char* in_dbname, Oid dboid, const char* username, Oid useroid) { m_indbname = in_dbname; m_dboid = dboid; m_username = username; + m_useroid = useroid; } void PostgresInitializer::InitBootstrap() @@ -1489,12 +1492,19 @@ void PostgresInitializer::InitSession() StartXact(); - if (IsUnderPostmaster) { - CheckAuthentication(); - InitUser(); - } else { + if (!IsUnderPostmaster) { CheckAtLeastOneRoles(); SetSuperUserStandalone(); + } else if (IsBackgroundWorker) { + if (m_username == NULL && !OidIsValid(m_useroid)) { + InitializeSessionUserIdStandalone(); + m_isSuperUser = true; + } else { + InitUser(); + } + } else { + CheckAuthentication(); + InitUser(); } CheckConnPermission(); @@ -1626,7 +1636,7 @@ void PostgresInitializer::SetSuperUserAndDatabase() void PostgresInitializer::InitUser() { - InitializeSessionUserId(m_username); + InitializeSessionUserId(m_username, m_useroid); m_isSuperUser = superuser(); u_sess->misc_cxt.CurrentUserName = u_sess->proc_cxt.MyProcPort->user_name; } diff --git a/src/common/backend/utils/misc/guc.cpp b/src/common/backend/utils/misc/guc.cpp index e0d216dea..a7977f878 100644 --- a/src/common/backend/utils/misc/guc.cpp +++ b/src/common/backend/utils/misc/guc.cpp @@ -459,6 +459,7 @@ static void assign_statistics_memory(int newval, void* extra); static void assign_history_memory(int newval, void* extra); static bool check_history_memory_limit(int* newval, void** extra, GucSource source); static bool check_autovacuum_max_workers(int* newval, void** extra, GucSource source); +static bool check_max_worker_processes(int* newval, void** extra, GucSource source); static bool check_job_max_workers(int* newval, void** extra, GucSource source); static bool check_effective_io_concurrency(int* newval, void** extra, GucSource source); static void assign_effective_io_concurrency(int newval, void* extra); @@ -7070,6 +7071,23 @@ static void init_configure_names_int() NULL, NULL }, + { + /* see max_connections */ + { + "max_background_workers", + PGC_POSTMASTER, + RESOURCES_ASYNCHRONOUS, + gettext_noop("Maximum number of concurrent background worker processes."), + NULL + }, + &g_instance.attr.attr_storage.max_background_workers, + 8, + 0, + MAX_BACKENDS, + check_max_worker_processes, + NULL, + NULL + }, { { "job_queue_processes", @@ -18748,6 +18766,7 @@ static bool check_maxconnections(int* newval, void** extra, GucSource source) } #endif if (*newval + g_instance.attr.attr_storage.autovacuum_max_workers + g_instance.attr.attr_sql.job_queue_processes + + g_instance.attr.attr_storage.max_background_workers + AUXILIARY_BACKENDS + AV_LAUNCHER_PROCS + g_instance.attr.attr_network.maxInnerToolConnections > MAX_BACKENDS) { return false; @@ -18758,6 +18777,7 @@ static bool check_maxconnections(int* newval, void** extra, GucSource source) static bool CheckMaxInnerToolConnections(int* newval, void** extra, GucSource source) { if (*newval + g_instance.attr.attr_storage.autovacuum_max_workers + g_instance.attr.attr_sql.job_queue_processes + + g_instance.attr.attr_storage.max_background_workers + g_instance.attr.attr_network.MaxConnections + AUXILIARY_BACKENDS + AV_LAUNCHER_PROCS > MAX_BACKENDS) { return false; } @@ -18767,6 +18787,7 @@ static bool CheckMaxInnerToolConnections(int* newval, void** extra, GucSource so static bool check_autovacuum_max_workers(int* newval, void** extra, GucSource source) { if (g_instance.attr.attr_network.MaxConnections + *newval + g_instance.attr.attr_sql.job_queue_processes + + g_instance.attr.attr_storage.max_background_workers + AUXILIARY_BACKENDS + AV_LAUNCHER_PROCS + g_instance.attr.attr_network.maxInnerToolConnections > MAX_BACKENDS) { return false; @@ -18774,6 +18795,18 @@ static bool check_autovacuum_max_workers(int* newval, void** extra, GucSource so return true; } +static bool check_max_worker_processes(int* newval, void** extra, GucSource source) +{ + if (g_instance.attr.attr_network.MaxConnections + g_instance.attr.attr_storage.autovacuum_max_workers + + g_instance.attr.attr_sql.job_queue_processes + *newval + + AUXILIARY_BACKENDS + AV_LAUNCHER_PROCS + g_instance.attr.attr_network.maxInnerToolConnections > + MAX_BACKENDS) { + return false; + } + return true; +} + + /* * Description: Check wheth out of max backends after max job worker threads. * @@ -18784,6 +18817,7 @@ static bool check_autovacuum_max_workers(int* newval, void** extra, GucSource so static bool check_job_max_workers(int* newval, void** extra, GucSource source) { if (g_instance.attr.attr_network.MaxConnections + g_instance.attr.attr_storage.autovacuum_max_workers + *newval + + g_instance.attr.attr_storage.max_background_workers + AUXILIARY_BACKENDS + AV_LAUNCHER_PROCS + g_instance.attr.attr_network.maxInnerToolConnections > MAX_BACKENDS) { return false; diff --git a/src/gausskernel/optimizer/commands/async.cpp b/src/gausskernel/optimizer/commands/async.cpp index 4668be049..e04af32a1 100755 --- a/src/gausskernel/optimizer/commands/async.cpp +++ b/src/gausskernel/optimizer/commands/async.cpp @@ -200,8 +200,6 @@ typedef struct QueueBackendStatus { QueuePosition pos; /* backend has read queue up to here */ } QueueBackendStatus; -#define InvalidPid ((ThreadId)(-1)) - /* * Shared memory state for LISTEN/NOTIFY (excluding its SLRU stuff) * diff --git a/src/gausskernel/process/postmaster/Makefile b/src/gausskernel/process/postmaster/Makefile index b2eb420f7..a491ec904 100755 --- a/src/gausskernel/process/postmaster/Makefile +++ b/src/gausskernel/process/postmaster/Makefile @@ -32,7 +32,7 @@ ifneq "$(MAKECMDGOALS)" "clean" endif endif OBJS = autovacuum.o bgwriter.o fork_process.o pgarch.o pgstat.o postmaster.o gaussdb_version.o\ - startup.o syslogger.o walwriter.o checkpointer.o pgaudit.o alarmchecker.o \ + startup.o syslogger.o walwriter.o checkpointer.o pgaudit.o alarmchecker.o bgworker.o\ twophasecleaner.o aiocompleter.o fencedudf.o lwlockmonitor.o cbmwriter.o remoteservice.o pagewriter.o\ $(top_builddir)/src/lib/config/libconfig.a diff --git a/src/gausskernel/process/postmaster/bgworker.cpp b/src/gausskernel/process/postmaster/bgworker.cpp new file mode 100644 index 000000000..580bf35a5 --- /dev/null +++ b/src/gausskernel/process/postmaster/bgworker.cpp @@ -0,0 +1,1309 @@ +/* -------------------------------------------------------------------- + * bgworker.cpp + * POSTGRES pluggable background workers implementation + * + * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/gausskernel/process/postmaster/bgworker.cpp + * + * ------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include + +#include "libpq/pqsignal.h" +#include "miscadmin.h" +#include "pgstat.h" +#include "postmaster/bgworker_internals.h" +#include "postmaster/postmaster.h" +#include "storage/ipc.h" +#include "storage/latch.h" +#include "storage/lwlock.h" +#include "storage/pg_shmem.h" +#include "storage/pmsignal.h" +#include "storage/proc.h" +#include "storage/procsignal.h" +#include "storage/shmem.h" +#include "tcop/tcopprot.h" +#include "utils/ascii.h" +#include "utils/ps_status.h" +#include "utils/postinit.h" + +/* + * The postmaster's list of registered background workers, in private memory. + */ +THR_LOCAL slist_head BackgroundWorkerList = SLIST_STATIC_INIT(BackgroundWorkerList); + +/* + * BackgroundWorkerSlots exist in shared memory and can be accessed (via + * the BackgroundWorkerArray) by both the postmaster and by regular backends. + * However, the postmaster cannot take locks, even spinlocks, because this + * might allow it to crash or become wedged if shared memory gets corrupted. + * Such an outcome is intolerable. Therefore, we need a lockless protocol + * for coordinating access to this data. + * + * The 'in_use' flag is used to hand off responsibility for the slot between + * the postmaster and the rest of the system. When 'in_use' is false, + * the postmaster will ignore the slot entirely, except for the 'in_use' flag + * itself, which it may read. In this state, regular backends may modify the + * slot. Once a backend sets 'in_use' to true, the slot becomes the + * responsibility of the postmaster. Regular backends may no longer modify it, + * but the postmaster may examine it. Thus, a backend initializing a slot + * must fully initialize the slot - and insert a write memory barrier - before + * marking it as in use. + * + * As an exception, however, even when the slot is in use, regular backends + * may set the 'terminate' flag for a slot, telling the postmaster not + * to restart it. Once the background worker is no longer running, the slot + * will be released for reuse. + * + * In addition to coordinating with the postmaster, backends modifying this + * data structure must coordinate with each other. Since they can take locks, + * this is straightforward: any backend wishing to manipulate a slot must + * take BackgroundWorkerLock in exclusive mode. Backends wishing to read + * data that might get concurrently modified by other backends should take + * this lock in shared mode. No matter what, backends reading this data + * structure must be able to tolerate concurrent modifications by the + * postmaster. + */ +typedef struct BackgroundWorkerSlot { + bool in_use; + bool terminate; + ThreadId pid; /* InvalidPid = not started yet; 0 = dead */ + uint64 generation; /* incremented when slot is recycled */ + BackgroundWorker worker; +} BackgroundWorkerSlot; + +/* + * In order to limit the total number of parallel workers (according to + * max_parallel_workers GUC), we maintain the number of active parallel + * workers. Since the postmaster cannot take locks, two variables are used for + * this purpose: the number of registered parallel workers (modified by the + * backends, protected by BackgroundWorkerLock) and the number of terminated + * parallel workers (modified only by the postmaster, lockless). The active + * number of parallel workers is the number of registered workers minus the + * terminated ones. These counters can of course overflow, but it's not + * important here since the subtraction will still give the right number. + */ +typedef struct BackgroundWorkerArray { + int total_slots; + uint32 parallel_register_count; // For extension only + uint32 parallel_terminate_count; // For extension only + BackgroundWorkerSlot slot[FLEXIBLE_ARRAY_MEMBER]; +} BackgroundWorkerArray; + +struct BackgroundWorkerHandle { + int slot; + uint64 generation; +}; + +/* + * List of internal background worker entry points. We need this for + * reasons explained in LookupBackgroundWorkerFunction(), below. + */ +static const struct { + const char *fn_name; + bgworker_main_type fn_addr; +} InternalBGWorkers[] = + +{ +}; + +/* Private functions. */ +static bgworker_main_type LookupBackgroundWorkerFunction(const char *libraryname, const char *funcname); + +/* + * Calculate shared memory needed. + */ +Size BackgroundWorkerShmemSize(void) +{ + Size size; + + /* Array of workers is variably sized. */ + size = offsetof(BackgroundWorkerArray, slot); + size = add_size(size, mul_size((Size)g_instance.attr.attr_storage.max_background_workers, + sizeof(BackgroundWorkerSlot))); + + return size; +} + +/* + * Initialize shared memory. + */ +void BackgroundWorkerShmemInit(void) +{ + bool found; + + t_thrd.bgworker_cxt.background_worker_data = (BackgroundWorkerArray*)ShmemInitStruct("Background Worker Data", + BackgroundWorkerShmemSize(), + &found); + if (!IsUnderPostmaster) { + slist_iter siter; + int slotno = 0; + + t_thrd.bgworker_cxt.background_worker_data->total_slots = g_instance.attr.attr_storage.max_background_workers; + t_thrd.bgworker_cxt.background_worker_data->parallel_register_count = 0; + t_thrd.bgworker_cxt.background_worker_data->parallel_terminate_count = 0; + + /* + * Copy contents of worker list into shared memory. Record the shared + * memory slot assigned to each worker. This ensures a 1-to-1 + * correspondence between the postmaster's private list and the array + * in shared memory. + */ + slist_foreach(siter, &BackgroundWorkerList) { + BackgroundWorkerSlot *slot = &t_thrd.bgworker_cxt.background_worker_data->slot[slotno]; + RegisteredBgWorker *rw; + + rw = slist_container(RegisteredBgWorker, rw_lnode, siter.cur); + Assert(slotno < g_instance.attr.attr_storage.max_background_workers); + slot->in_use = true; + slot->terminate = false; + slot->pid = InvalidPid; + slot->generation = 0; + rw->rw_shmem_slot = slotno; + rw->rw_worker.bgw_notify_pid = 0; /* might be reinit after crash */ + int ss_rc = memcpy_s(&slot->worker, sizeof(BackgroundWorker), &rw->rw_worker, sizeof(BackgroundWorker)); + securec_check(ss_rc, "\0", "\0"); + ++slotno; + } + + /* + * Mark any remaining slots as not in use. + */ + while (slotno < g_instance.attr.attr_storage.max_background_workers) { + BackgroundWorkerSlot *slot = &t_thrd.bgworker_cxt.background_worker_data->slot[slotno]; + + slot->in_use = false; + ++slotno; + } + } else { + Assert(found); + } +} + +/* + * Search the postmaster's backend-private list of RegisteredBgWorker objects + * for the one that maps to the given slot number. + */ +static RegisteredBgWorker * FindRegisteredWorkerBySlotNumber(int slotno) +{ + slist_iter siter; + + slist_foreach(siter, &BackgroundWorkerList) { + RegisteredBgWorker *rw = slist_container(RegisteredBgWorker, rw_lnode, siter.cur); + if (rw->rw_shmem_slot == slotno) { + return rw; + } + } + + return NULL; +} + +/* + * Notice changes to shared memory made by other backends. This code + * runs in the postmaster, so we must be very careful not to assume that + * shared memory contents are sane. Otherwise, a rogue backend could take + * out the postmaster. + */ +void BackgroundWorkerStateChange(void) +{ + int slotno; + + /* + * The total number of slots stored in shared memory should match our + * notion of max_background_workers. If it does not, something is very + * wrong. Further down, we always refer to this value as + * max_background_workers, in case shared memory gets corrupted while we're + * looping. + */ + if (g_instance.attr.attr_storage.max_background_workers != t_thrd.bgworker_cxt.background_worker_data->total_slots) { + elog(LOG, + "inconsistent background worker state (max_background_workers=%d, total_slots=%d", + g_instance.attr.attr_storage.max_background_workers, + t_thrd.bgworker_cxt.background_worker_data->total_slots); + return; + } + + /* + * Iterate through slots, looking for newly-registered workers or workers + * who must die. + */ + for (slotno = 0; slotno < g_instance.attr.attr_storage.max_background_workers; ++slotno) { + BackgroundWorkerSlot *slot = &t_thrd.bgworker_cxt.background_worker_data->slot[slotno]; + RegisteredBgWorker *rw = NULL; + + if (!slot->in_use) { + continue; + } + + /* + * Make sure we don't see the in_use flag before the updated slot + * contents. + */ + pg_read_barrier(); + + /* See whether we already know about this worker. */ + rw = FindRegisteredWorkerBySlotNumber(slotno); + if (rw != NULL) { + /* + * In general, the worker data can't change after it's initially + * registered. However, someone can set the terminate flag. + */ + if (slot->terminate && !rw->rw_terminate) { + rw->rw_terminate = true; + if (rw->rw_pid != 0) { + if (gs_signal_send(rw->rw_pid, SIGTERM) != 0) { + ereport(WARNING, + (errmsg("sending SIGTERM to %lu failed", rw->rw_pid))); + } + } else { + /* Report never-started, now-terminated worker as dead. */ + ReportBackgroundWorkerPID(rw); + } + } + continue; + } + + /* + * If the worker is marked for termination, we don't need to add it to + * the registered workers list; we can just free the slot. However, if + * bgw_notify_pid is set, the process that registered the worker may + * need to know that we've processed the terminate request, so be sure + * to signal it. + */ + if (slot->terminate) { + /* + * We need a memory barrier here to make sure that the load of + * bgw_notify_pid and the update of parallel_terminate_count + * complete before the store to in_use. + */ + ThreadId notify_pid = slot->worker.bgw_notify_pid; + if ((slot->worker.bgw_flags & BGWORKER_CLASS_PARALLEL) != 0) { + t_thrd.bgworker_cxt.background_worker_data->parallel_terminate_count++; + } + pg_memory_barrier(); + slot->pid = 0; + slot->in_use = false; + if (notify_pid != 0) { + if (gs_signal_send(notify_pid, SIGUSR1) != 0) { + ereport(WARNING, + (errmsg("sending SIGUSR1 to %lu failed", notify_pid))); + } + } + + continue; + } + + /* + * Copy the registration data into the registered workers list. + */ + rw = (RegisteredBgWorker*)malloc(sizeof(RegisteredBgWorker)); + if (rw == NULL) { + ereport(LOG, + (errcode(ERRCODE_OUT_OF_MEMORY), + errmsg("out of memory"))); + return; + } + + /* + * Copy strings in a paranoid way. If shared memory is corrupted, the + * source data might not even be NUL-terminated. + */ + ascii_safe_strlcpy(rw->rw_worker.bgw_name, + slot->worker.bgw_name, BGW_MAXLEN); + ascii_safe_strlcpy(rw->rw_worker.bgw_type, + slot->worker.bgw_type, BGW_MAXLEN); + ascii_safe_strlcpy(rw->rw_worker.bgw_library_name, + slot->worker.bgw_library_name, BGW_MAXLEN); + ascii_safe_strlcpy(rw->rw_worker.bgw_function_name, + slot->worker.bgw_function_name, BGW_MAXLEN); + + /* + * Copy various fixed-size fields. + * + * flags, start_time, and restart_time are examined by the postmaster, + * but nothing too bad will happen if they are corrupted. The + * remaining fields will only be examined by the child process. It + * might crash, but we won't. + */ + rw->rw_worker.bgw_flags = slot->worker.bgw_flags; + rw->rw_worker.bgw_start_time = slot->worker.bgw_start_time; + rw->rw_worker.bgw_restart_time = slot->worker.bgw_restart_time; + rw->rw_worker.bgw_main_arg = slot->worker.bgw_main_arg; + int ss_rc = memcpy_s(rw->rw_worker.bgw_extra, BGW_EXTRALEN, slot->worker.bgw_extra, BGW_EXTRALEN); + securec_check(ss_rc, "\0", "\0"); + + /* + * Copy the PID to be notified about state changes, but only if the + * postmaster knows about a backend with that PID. It isn't an error + * if the postmaster doesn't know about the PID, because the backend + * that requested the worker could have died (or been killed) just + * after doing so. Nonetheless, at least until we get some experience + * with how this plays out in the wild, log a message at a relative + * high debug level. + */ + rw->rw_worker.bgw_notify_pid = slot->worker.bgw_notify_pid; + if (!PostmasterMarkPIDForWorkerNotify(rw->rw_worker.bgw_notify_pid)) { + elog(DEBUG1, "worker notification PID %lu is not valid", + rw->rw_worker.bgw_notify_pid); + rw->rw_worker.bgw_notify_pid = 0; + } + + /* Initialize postmaster bookkeeping. */ + rw->rw_backend = NULL; + rw->rw_pid = 0; + rw->rw_child_slot = 0; + rw->rw_crashed_at = 0; + rw->rw_shmem_slot = slotno; + rw->rw_terminate = false; + + /* Log it! */ + ereport(DEBUG1, + (errmsg("registering background worker \"%s\"", + rw->rw_worker.bgw_name))); + + slist_push_head(&BackgroundWorkerList, &rw->rw_lnode); + } +} + +/* + * Forget about a background worker that's no longer needed. + * + * The worker must be identified by passing an slist_mutable_iter that + * points to it. This convention allows deletion of workers during + * searches of the worker list, and saves having to search the list again. + * + * This function must be invoked only in the postmaster. + */ +void ForgetBackgroundWorker(slist_mutable_iter *cur) +{ + RegisteredBgWorker *rw = NULL; + BackgroundWorkerSlot *slot = NULL; + + rw = slist_container(RegisteredBgWorker, rw_lnode, cur->cur); + + Assert(rw->rw_shmem_slot < g_instance.attr.attr_storage.max_background_workers); + slot = &t_thrd.bgworker_cxt.background_worker_data->slot[rw->rw_shmem_slot]; + if ((rw->rw_worker.bgw_flags & BGWORKER_CLASS_PARALLEL) != 0) { + t_thrd.bgworker_cxt.background_worker_data->parallel_terminate_count++; + } + + slot->in_use = false; + + ereport(DEBUG1, + (errmsg("unregistering background worker \"%s\"", + rw->rw_worker.bgw_name))); + + slist_delete_current(cur); + free(rw); +} + +/* + * Report the PID of a newly-launched background worker in shared memory. + * + * This function should only be called from the postmaster. + */ +void ReportBackgroundWorkerPID(const RegisteredBgWorker *rw) +{ + BackgroundWorkerSlot *slot; + + Assert(rw->rw_shmem_slot < g_instance.attr.attr_storage.max_background_workers); + slot = &t_thrd.bgworker_cxt.background_worker_data->slot[rw->rw_shmem_slot]; + slot->pid = rw->rw_pid; + ereport(LOG, + (errmsg("ReportBackgroundWorkerPID slot: %d, pid: %lu, bgw_notify_pid: %lu", + rw->rw_shmem_slot, slot->pid, rw->rw_worker.bgw_notify_pid))); + + if (rw->rw_worker.bgw_notify_pid != 0) { + int ret = gs_signal_send(rw->rw_worker.bgw_notify_pid, SIGUSR1); + ereport(LOG, + (errmsg("ReportBackgroundWorkerPID send SIGUSR1 to bgw_notify_pid: %lu, ret: %d", + rw->rw_worker.bgw_notify_pid, ret))); + } +} + +/* + * Report that the PID of a background worker is now zero because a + * previously-running background worker has exited. + * + * This function should only be called from the postmaster. + */ +void ReportBackgroundWorkerExit(slist_mutable_iter *cur) +{ + RegisteredBgWorker *rw = slist_container(RegisteredBgWorker, rw_lnode, cur->cur); + + Assert(rw->rw_shmem_slot < g_instance.attr.attr_storage.max_background_workers); + BackgroundWorkerSlot *slot = &t_thrd.bgworker_cxt.background_worker_data->slot[rw->rw_shmem_slot]; + slot->pid = rw->rw_pid; + ThreadId notify_pid = rw->rw_worker.bgw_notify_pid; + + /* + * If this worker is slated for deregistration, do that before notifying + * the process which started it. Otherwise, if that process tries to + * reuse the slot immediately, it might not be available yet. In theory + * that could happen anyway if the process checks slot->pid at just the + * wrong moment, but this makes the window narrower. + */ + if (rw->rw_terminate || + rw->rw_worker.bgw_restart_time == BGW_NEVER_RESTART) { + ForgetBackgroundWorker(cur); + } + + if (notify_pid != 0) { + int ret = gs_signal_send(notify_pid, SIGUSR1); + ereport(LOG, + (errmsg("ReportBackgroundWorkerExit send SIGUSR1 to bgw_notify_pid: %lu, ret: %d", + notify_pid, ret))); + } +} + +/* + * Cancel SIGUSR1 notifications for a PID belonging to an exiting backend. + * + * This function should only be called from the postmaster. + */ +void BackgroundWorkerStopNotifications(ThreadId pid) +{ + slist_iter siter; + + slist_foreach(siter, &BackgroundWorkerList) + { + RegisteredBgWorker *rw = slist_container(RegisteredBgWorker, rw_lnode, siter.cur); + if (rw->rw_worker.bgw_notify_pid == pid) { + rw->rw_worker.bgw_notify_pid = 0; + } + } +} + +/* + * Reset background worker crash state. + * + * We assume that, after a crash-and-restart cycle, background workers without + * the never-restart flag should be restarted immediately, instead of waiting + * for bgw_restart_time to elapse. + */ +void ResetBackgroundWorkerCrashTimes(void) +{ + slist_mutable_iter iter; + + slist_foreach_modify(iter, &BackgroundWorkerList) + { + RegisteredBgWorker *rw = slist_container(RegisteredBgWorker, rw_lnode, iter.cur); + + if (rw->rw_worker.bgw_restart_time == BGW_NEVER_RESTART) { + /* + * Workers marked BGW_NEVER_RESTART shouldn't get relaunched after + * the crash, so forget about them. (If we wait until after the + * crash to forget about them, and they are parallel workers, + * parallel_terminate_count will get incremented after we've + * already zeroed parallel_register_count, which would be bad.) + */ + ForgetBackgroundWorker(&iter); + } else { + /* + * The accounting which we do via parallel_register_count and + * parallel_terminate_count would get messed up if a worker marked + * parallel could survive a crash and restart cycle. All such + * workers should be marked BGW_NEVER_RESTART, and thus control + * should never reach this branch. + */ + Assert((rw->rw_worker.bgw_flags & BGWORKER_CLASS_PARALLEL) == 0); + + /* + * Allow this worker to be restarted immediately after we finish + * resetting. + */ + rw->rw_crashed_at = 0; + } + } +} + +#ifdef EXEC_BACKEND +/* + * In EXEC_BACKEND mode, return address of the corresponding slot in + * shared memory. + */ +void* GetBackgroundWorkerShmAddr(int slotno) +{ + Assert(slotno < t_thrd.bgworker_cxt.background_worker_data->total_slots); + return (void*)&t_thrd.bgworker_cxt.background_worker_data->slot[slotno]; +} + +/* + * In EXEC_BACKEND mode, workers use this to retrieve their details from + * shared memory. + */ +BackgroundWorker* BackgroundWorkerEntry(const BackgroundWorkerSlot* bgWorkerSlotShmAddr) +{ + static THR_LOCAL BackgroundWorker myEntry; + + Assert(bgWorkerSlotShmAddr != NULL); + Assert(bgWorkerSlotShmAddr->in_use); + + /* must copy this in case we don't intend to retain shmem access */ + int ss_rc = memcpy_s(&myEntry, sizeof(myEntry), &bgWorkerSlotShmAddr->worker, sizeof(myEntry)); + securec_check(ss_rc, "\0", "\0"); + return &myEntry; +} +#endif + +/* + * Complain about the BackgroundWorker definition using error level elevel. + * Return true if it looks ok, false if not (unless elevel >= ERROR, in + * which case we won't return at all in the not-OK case). + */ +static bool SanityCheckBackgroundWorker(BackgroundWorker *worker, int elevel) +{ + /* sanity check for flags */ + if (worker->bgw_flags & BGWORKER_BACKEND_DATABASE_CONNECTION) { + if (!(worker->bgw_flags & BGWORKER_SHMEM_ACCESS)) { + ereport(elevel, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("background worker \"%s\": must attach to shared memory in order to request a database connection", + worker->bgw_name))); + return false; + } + + if (worker->bgw_start_time == BgWorkerStart_PostmasterStart) { + ereport(elevel, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("background worker \"%s\": cannot request database access if starting at postmaster start", + worker->bgw_name))); + return false; + } + + /* XXX other checks? */ + } + + if ((worker->bgw_restart_time < 0 && + worker->bgw_restart_time != BGW_NEVER_RESTART) || + (worker->bgw_restart_time > USECS_PER_DAY / 1000)) { + ereport(elevel, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("background worker \"%s\": invalid restart interval", + worker->bgw_name))); + return false; + } + + /* + * Parallel workers may not be configured for restart, because the + * parallel_register_count/parallel_terminate_count accounting can't + * handle parallel workers lasting through a crash-and-restart cycle. + */ + if (worker->bgw_restart_time != BGW_NEVER_RESTART && + (worker->bgw_flags & BGWORKER_CLASS_PARALLEL) != 0) { + ereport(elevel, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("background worker \"%s\": parallel workers may not be configured for restart", + worker->bgw_name))); + return false; + } + + /* + * If bgw_type is not filled in, use bgw_name. + */ + if (strcmp(worker->bgw_type, "") == 0) { + int rd = strncpy_s(worker->bgw_type, BGW_MAXLEN, worker->bgw_name, BGW_MAXLEN); + securec_check(rd, "\0", "\0"); + } + + return true; +} + +static void bgworker_quickdie(SIGNAL_ARGS) +{ + /* + * We DO NOT want to run proc_exit() or atexit() callbacks -- we're here + * because shared memory may be corrupted, so we don't want to try to + * clean up our transaction. Just nail the windows shut and get out of + * town. The callbacks wouldn't be safe to run from a signal handler, + * anyway. + * + * Note we do _exit(2) not _exit(0). This is to force the postmaster into + * a system reset cycle if someone sends a manual SIGQUIT to a random + * backend. This is necessary precisely because we don't clean up our + * shared memory state. (The "dead man switch" mechanism in pmsignal.c + * should ensure the postmaster sees this as a crash, too, but no harm in + * being doubly sure.) + */ + _exit(2); +} + +/* + * Standard SIGTERM handler for background workers + */ +static void bgworker_die(SIGNAL_ARGS) +{ + (void)PG_SETMASK(&t_thrd.libpq_cxt.BlockSig); + + ereport(FATAL, + (errcode(ERRCODE_ADMIN_SHUTDOWN), + errmsg("terminating background worker \"%s\" due to administrator command", + t_thrd.bgworker_cxt.my_bgworker_entry->bgw_type))); +} + +/* + * Standard SIGUSR1 handler for unconnected workers + * + * Here, we want to make sure an unconnected worker will at least heed + * latch activity. + */ +static void bgworker_sigusr1_handler(SIGNAL_ARGS) +{ + int save_errno = errno; + + latch_sigusr1_handler(); + + errno = save_errno; +} + +/* + * Start a new background worker + * + * This is the main entry point for background worker, to be called from + * postmaster. + */ +void StartBackgroundWorker(void* bgWorkerSlotShmAddr) +{ + sigjmp_buf local_sigjmp_buf; + t_thrd.bgworker_cxt.my_bgworker_entry = BackgroundWorkerEntry((BackgroundWorkerSlot *)bgWorkerSlotShmAddr); + BackgroundWorker *worker = t_thrd.bgworker_cxt.my_bgworker_entry; + bgworker_main_type entrypt; + + /* + * Create memory context and buffer used for RowDescription messages. As + * SendRowDescriptionMessage(), via exec_describe_statement_message(), is + * frequently executed for ever single statement, we don't want to + * allocate a separate buffer every time. + */ + t_thrd.mem_cxt.row_desc_mem_cxt = AllocSetContextCreate(t_thrd.top_mem_cxt, + "RowDescriptionContext", + ALLOCSET_DEFAULT_MINSIZE, + ALLOCSET_DEFAULT_INITSIZE, + ALLOCSET_DEFAULT_MAXSIZE); + MemoryContext old_mc = MemoryContextSwitchTo(t_thrd.mem_cxt.row_desc_mem_cxt); + initStringInfo(&(*t_thrd.postgres_cxt.row_description_buf)); + (void)MemoryContextSwitchTo(old_mc); + + t_thrd.mem_cxt.mask_password_mem_cxt = AllocSetContextCreate(t_thrd.top_mem_cxt, + "MaskPasswordCtx", + ALLOCSET_DEFAULT_MINSIZE, + ALLOCSET_DEFAULT_INITSIZE, + ALLOCSET_DEFAULT_MAXSIZE); + + if (worker == NULL) { + ereport(FATAL, + (errmsg("unable to find bgworker entry"))); + } + + IsBackgroundWorker = true; + + /* Identify myself via ps */ + init_ps_display(worker->bgw_name, "", "", ""); + + SetProcessingMode(InitProcessing); + + /* + * Set up signal handlers. + */ + if (worker->bgw_flags & BGWORKER_BACKEND_DATABASE_CONNECTION) { + /* + * SIGINT is used to signal canceling the current action + */ + (void)gspqsignal(SIGINT, StatementCancelHandler); + (void)gspqsignal(SIGUSR1, procsignal_sigusr1_handler); + (void)gspqsignal(SIGFPE, FloatExceptionHandler); + + /* XXX Any other handlers needed here? */ + } else { + (void)gspqsignal(SIGINT, SIG_IGN); + (void)gspqsignal(SIGUSR1, bgworker_sigusr1_handler); + (void)gspqsignal(SIGFPE, SIG_IGN); + } + (void)gspqsignal(SIGTERM, bgworker_die); + (void)gspqsignal(SIGHUP, SIG_IGN); + + (void)gspqsignal(SIGQUIT, bgworker_quickdie); + (void)gspqsignal(SIGALRM, handle_sig_alarm); + + (void)gspqsignal(SIGPIPE, SIG_IGN); + (void)gspqsignal(SIGUSR2, SIG_IGN); + (void)gspqsignal(SIGCHLD, SIG_DFL); + + /* + * If an exception is encountered, processing resumes here. + * + * See notes in postgres.c about the design of this coding. + */ + if (sigsetjmp(local_sigjmp_buf, 1) != 0) { + /* Since not using PG_TRY, must reset error stack by hand */ + t_thrd.log_cxt.error_context_stack = NULL; + + /* Prevent interrupts while cleaning up */ + HOLD_INTERRUPTS(); + + /* Report the error to the server log */ + EmitErrorReport(); + + /* + * Do we need more cleanup here? For shmem-connected bgworkers, we + * will call InitProcess below, which will install ProcKill as exit + * callback. That will take care of releasing locks, etc. + */ + + /* and go away */ + proc_exit(1); + } + + /* We can now handle ereport(ERROR) */ + t_thrd.log_cxt.PG_exception_stack = &local_sigjmp_buf; + + /* + * If the background worker request shared memory access, set that up now; + * else, detach all shared memory segments. + */ + if (worker->bgw_flags & BGWORKER_SHMEM_ACCESS) { + /* + * Early initialization. Some of this could be useful even for + * background workers that aren't using shared memory, but they can + * call the individual startup routines for those subsystems if + * needed. + */ + BaseInit(); + + /* + * Create a per-backend PGPROC struct in shared memory, except in the + * EXEC_BACKEND case where this was done in SubPostmasterMain. We must + * do this before we can use LWLocks (and in the EXEC_BACKEND case we + * already had to do some stuff with LWLocks). + */ +#ifndef EXEC_BACKEND + InitProcess(); +#endif + } + + /* + * Look up the entry point function, loading its library if necessary. + */ + entrypt = LookupBackgroundWorkerFunction(worker->bgw_library_name, + worker->bgw_function_name); + + /* + * Note that in normal processes, we would call InitPostgres here. For a + * worker, however, we don't know what database to connect to, yet; so we + * need to wait until the user code does it via + * BackgroundWorkerInitializeConnection(). + */ + + /* + * Now invoke the user-defined worker code + */ + entrypt(worker->bgw_main_arg); + + /* ... and if it returns, we're done */ + proc_exit(0); +} + +/* + * Register a new static background worker. + * + * This can only be called directly from postmaster or in the _PG_init + * function of a module library that's loaded by shared_preload_libraries; + * otherwise it will have no effect. + */ +void RegisterBackgroundWorker(BackgroundWorker *worker) +{ + RegisteredBgWorker *rw; + static THR_LOCAL int numworkers = 0; + + if (!IsUnderPostmaster) { + ereport(DEBUG1, + (errmsg("registering background worker \"%s\"", worker->bgw_name))); + } + + if (!u_sess->misc_cxt.process_shared_preload_libraries_in_progress && + strcmp(worker->bgw_library_name, "postgres") != 0) { + if (!IsUnderPostmaster) { + ereport(LOG, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("background worker \"%s\": must be registered in shared_preload_libraries", + worker->bgw_name))); + } + return; + } + + if (!SanityCheckBackgroundWorker(worker, LOG)) { + return; + } + + if (worker->bgw_notify_pid != 0) { + ereport(LOG, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("background worker \"%s\": only dynamic background workers can request notification", + worker->bgw_name))); + return; + } + + /* + * Enforce maximum number of workers. Note this is overly restrictive: we + * could allow more non-shmem-connected workers, because these don't count + * towards the MAX_BACKENDS limit elsewhere. For now, it doesn't seem + * important to relax this restriction. + */ + if (++numworkers > g_instance.attr.attr_storage.max_background_workers) { + ereport(LOG, + (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED), + errmsg("too many background workers"), + errdetail_plural("Up to %d background worker can be registered with the current settings.", + "Up to %d background workers can be registered with the current settings.", + g_instance.attr.attr_storage.max_background_workers, + g_instance.attr.attr_storage.max_background_workers), + errhint("Consider increasing the configuration parameter \"max_background_workers\"."))); + return; + } + + /* + * Copy the registration data into the registered workers list. + */ + rw = (RegisteredBgWorker*)malloc(sizeof(RegisteredBgWorker)); + if (rw == NULL) { + ereport(LOG, + (errcode(ERRCODE_OUT_OF_MEMORY), + errmsg("out of memory"))); + return; + } + + rw->rw_worker = *worker; + rw->rw_backend = NULL; + rw->rw_pid = 0; + rw->rw_child_slot = 0; + rw->rw_crashed_at = 0; + rw->rw_terminate = false; + + slist_push_head(&BackgroundWorkerList, &rw->rw_lnode); +} + +/* + * Register a new background worker from a regular backend. + * + * Returns true on success and false on failure. Failure typically indicates + * that no background worker slots are currently available. + * + * If handle != NULL, we'll set *handle to a pointer that can subsequently + * be used as an argument to GetBackgroundWorkerPid(). The caller can + * free this pointer using pfree(), if desired. + */ +bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, + BackgroundWorkerHandle **handle) +{ + int slotno; + bool success = false; + bool parallel; + uint64 generation = 0; + + /* + * We can't register dynamic background workers from the postmaster. If + * this is a standalone backend, we're the only process and can't start + * any more. In a multi-process environment, it might be theoretically + * possible, but we don't currently support it due to locking + * considerations; see comments on the BackgroundWorkerSlot data + * structure. + */ + if (!IsUnderPostmaster) { + return false; + } + + if (!SanityCheckBackgroundWorker(worker, ERROR)) { + return false; + } + + parallel = (worker->bgw_flags & BGWORKER_CLASS_PARALLEL) != 0; + + (void)LWLockAcquire(BackgroundWorkerLock, LW_EXCLUSIVE); + + /* + * If this is a parallel worker, check whether there are already too many + * parallel workers; if so, don't register another one. Our view of + * parallel_terminate_count may be slightly stale, but that doesn't really + * matter: we would have gotten the same result if we'd arrived here + * slightly earlier anyway. There's no help for it, either, since the + * postmaster must not take locks; a memory barrier wouldn't guarantee + * anything useful. + */ + if (parallel && (int)(t_thrd.bgworker_cxt.background_worker_data->parallel_register_count - + t_thrd.bgworker_cxt.background_worker_data->parallel_terminate_count) >= + g_instance.shmem_cxt.max_parallel_workers) { + Assert(t_thrd.bgworker_cxt.background_worker_data->parallel_register_count - + t_thrd.bgworker_cxt.background_worker_data->parallel_terminate_count <= + MAX_PARALLEL_WORKER_LIMIT); + LWLockRelease(BackgroundWorkerLock); + return false; + } + + /* + * Look for an unused slot. If we find one, grab it. + */ + for (slotno = 0; slotno < t_thrd.bgworker_cxt.background_worker_data->total_slots; ++slotno) { + BackgroundWorkerSlot *slot = &t_thrd.bgworker_cxt.background_worker_data->slot[slotno]; + + if (!slot->in_use) { + int ss_rc = memcpy_s(&slot->worker, sizeof(BackgroundWorker), worker, sizeof(BackgroundWorker)); + securec_check(ss_rc, "\0", "\0"); + slot->pid = InvalidPid; /* indicates not started yet */ + slot->generation++; + slot->terminate = false; + generation = slot->generation; + if (parallel) + t_thrd.bgworker_cxt.background_worker_data->parallel_register_count++; + + /* + * Make sure postmaster doesn't see the slot as in use before it + * sees the new contents. + */ + pg_write_barrier(); + + slot->in_use = true; + success = true; + break; + } + } + + LWLockRelease(BackgroundWorkerLock); + + /* If we found a slot, tell the postmaster to notice the change. */ + if (success) { + SendPostmasterSignal(PMSIGNAL_BACKGROUND_WORKER_CHANGE); + } + + /* + * If we found a slot and the user has provided a handle, initialize it. + */ + if (success && handle) { + *handle = (BackgroundWorkerHandle*)palloc(sizeof(BackgroundWorkerHandle)); + (*handle)->slot = slotno; + (*handle)->generation = generation; + } + + return success; +} + +/* + * Get the PID of a dynamically-registered background worker. + * + * If the worker is determined to be running, the return value will be + * BGWH_STARTED and *pidp will get the PID of the worker process. If the + * postmaster has not yet attempted to start the worker, the return value will + * be BGWH_NOT_YET_STARTED. Otherwise, the return value is BGWH_STOPPED. + * + * BGWH_STOPPED can indicate either that the worker is temporarily stopped + * (because it is configured for automatic restart and exited non-zero), + * or that the worker is permanently stopped (because it exited with exit + * code 0, or was not configured for automatic restart), or even that the + * worker was unregistered without ever starting (either because startup + * failed and the worker is not configured for automatic restart, or because + * TerminateBackgroundWorker was used before the worker was successfully + * started). + */ +BgwHandleStatus GetBackgroundWorkerPid(const BackgroundWorkerHandle *handle, ThreadId *pidp) +{ + ThreadId pid = InvalidPid; + + Assert(handle->slot < g_instance.attr.attr_storage.max_background_workers); + BackgroundWorkerSlot* slot = &t_thrd.bgworker_cxt.background_worker_data->slot[handle->slot]; + + /* + * We could probably arrange to synchronize access to data using memory + * barriers only, but for now, let's just keep it simple and grab the + * lock. It seems unlikely that there will be enough traffic here to + * result in meaningful contention. + */ + (void)LWLockAcquire(BackgroundWorkerLock, LW_SHARED); + + /* + * The generation number can't be concurrently changed while we hold the + * lock. The pid, which is updated by the postmaster, can change at any + * time, but we assume such changes are atomic. So the value we read + * won't be garbage, but it might be out of date by the time the caller + * examines it (but that's unavoidable anyway). + * + * The in_use flag could be in the process of changing from true to false, + * but if it is already false then it can't change further. + */ + if (handle->generation != slot->generation || !slot->in_use) { + pid = 0; + } else { + pid = slot->pid; + } + + /* All done. */ + LWLockRelease(BackgroundWorkerLock); + + ereport(LOG, + (errmsg("GetBackgroundWorkerPid slot: %d, pid: %lu", + handle->slot, pid))); + if (pid == 0) { + return BGWH_STOPPED; + } else if (pid == InvalidPid) { + return BGWH_NOT_YET_STARTED; + } + *pidp = pid; + return BGWH_STARTED; +} + +/* + * Wait for a background worker to start up. + * + * This is like GetBackgroundWorkerPid(), except that if the worker has not + * yet started, we wait for it to do so; thus, BGWH_NOT_YET_STARTED is never + * returned. However, if the postmaster has died, we give up and return + * BGWH_POSTMASTER_DIED, since it that case we know that startup will not + * take place. + */ +BgwHandleStatus WaitForBackgroundWorkerStartup(const BackgroundWorkerHandle *handle, ThreadId *pidp) +{ + BgwHandleStatus status; + int rc; + volatile knl_thrd_context* localThrd = &t_thrd; + + for (;;) { + ThreadId pid = 0; + + CHECK_FOR_INTERRUPTS(); + + status = GetBackgroundWorkerPid(handle, &pid); + ereport(LOG, + (errmsg("WaitForBackgroundWorkerStartup slot: %d, pid: %lu, status: %u, mypid: %lu", + handle->slot, pid, status, t_thrd.proc_cxt.MyProcPid))); + ereport(LOG, + (errmsg("WaitForBackgroundWorkerStartup addr: %p", localThrd))); + if (status == BGWH_STARTED) { + *pidp = pid; + } + if (status != BGWH_NOT_YET_STARTED) { + break; + } + + rc = WaitLatch(&t_thrd.proc->procLatch, + WL_LATCH_SET | WL_POSTMASTER_DEATH, 0); + + if (rc & WL_POSTMASTER_DEATH) { + status = BGWH_POSTMASTER_DIED; + break; + } + + ResetLatch(&t_thrd.proc->procLatch); + } + + return status; +} + +/* + * Wait for a background worker to stop. + * + * If the worker hasn't yet started, or is running, we wait for it to stop + * and then return BGWH_STOPPED. However, if the postmaster has died, we give + * up and return BGWH_POSTMASTER_DIED, because it's the postmaster that + * notifies us when a worker's state changes. + */ +BgwHandleStatus WaitForBackgroundWorkerShutdown(const BackgroundWorkerHandle *handle) +{ + BgwHandleStatus status; + int rc; + + for (;;) { + ThreadId pid = InvalidPid; + + CHECK_FOR_INTERRUPTS(); + + status = GetBackgroundWorkerPid(handle, &pid); + if (status == BGWH_STOPPED) { + break; + } + + rc = WaitLatch(&t_thrd.proc->procLatch, + WL_LATCH_SET | WL_POSTMASTER_DEATH, 0); + + if (rc & WL_POSTMASTER_DEATH) { + status = BGWH_POSTMASTER_DIED; + break; + } + + ResetLatch(&t_thrd.proc->procLatch); + } + + return status; +} + +/* + * Instruct the postmaster to terminate a background worker. + * + * Note that it's safe to do this without regard to whether the worker is + * still running, or even if the worker may already have existed and been + * unregistered. + */ +void TerminateBackgroundWorker(const BackgroundWorkerHandle *handle) +{ + bool signal_postmaster = false; + + Assert(handle->slot < g_instance.attr.attr_storage.max_background_workers); + BackgroundWorkerSlot* slot = &t_thrd.bgworker_cxt.background_worker_data->slot[handle->slot]; + + /* Set terminate flag in shared memory, unless slot has been reused. */ + (void)LWLockAcquire(BackgroundWorkerLock, LW_EXCLUSIVE); + if (handle->generation == slot->generation) { + slot->terminate = true; + signal_postmaster = true; + } + LWLockRelease(BackgroundWorkerLock); + + /* Make sure the postmaster notices the change to shared memory. */ + if (signal_postmaster) { + SendPostmasterSignal(PMSIGNAL_BACKGROUND_WORKER_CHANGE); + } +} + +/* + * Look up (and possibly load) a bgworker entry point function. + * + * For functions contained in the core code, we use library name "postgres" + * and consult the InternalBGWorkers array. External functions are + * looked up, and loaded if necessary, using load_external_function(). + * + * The point of this is to pass function names as strings across process + * boundaries. We can't pass actual function addresses because of the + * possibility that the function has been loaded at a different address + * in a different process. This is obviously a hazard for functions in + * loadable libraries, but it can happen even for functions in the core code + * on platforms using EXEC_BACKEND (e.g., Windows). + * + * At some point it might be worthwhile to get rid of InternalBGWorkers[] + * in favor of applying load_external_function() for core functions too; + * but that raises portability issues that are not worth addressing now. + */ +static bgworker_main_type LookupBackgroundWorkerFunction(const char *libraryname, const char *funcname) +{ + /* + * If the function is to be loaded from postgres itself, search the + * InternalBGWorkers array. + */ + if (strcmp(libraryname, "postgres") == 0) { + size_t i; + for (i = 0; i < lengthof(InternalBGWorkers); i++) { + if (strcmp(InternalBGWorkers[i].fn_name, funcname) == 0) { + return InternalBGWorkers[i].fn_addr; + } + } + + /* We can only reach this by programming error. */ + elog(ERROR, "internal function \"%s\" not found", funcname); + } + + /* Otherwise load from external library. */ + return (bgworker_main_type) + load_external_function(libraryname, (char*)funcname, true, true).user_fn; +} + +/* + * Given a PID, get the bgw_type of the background worker. Returns NULL if + * not a valid background worker. + * + * The return value is in static memory belonging to this function, so it has + * to be used before calling this function again. This is so that the caller + * doesn't have to worry about the background worker locking protocol. + */ +const char * GetBackgroundWorkerTypeByPid(ThreadId pid) +{ + int slotno; + bool found = false; + static THR_LOCAL char result[BGW_MAXLEN]; + + (void)LWLockAcquire(BackgroundWorkerLock, LW_SHARED); + + for (slotno = 0; slotno < t_thrd.bgworker_cxt.background_worker_data->total_slots; slotno++) { + BackgroundWorkerSlot *slot = &t_thrd.bgworker_cxt.background_worker_data->slot[slotno]; + + if (slot->pid > 0 && slot->pid == pid) { + int rd = strncpy_s(result, BGW_MAXLEN, slot->worker.bgw_type, BGW_MAXLEN); + securec_check(rd, "\0", "\0"); + found = true; + break; + } + } + + LWLockRelease(BackgroundWorkerLock); + + if (!found) { + return NULL; + } + + return result; +} + +/* + * Connect background worker to a database. + */ +void BackgroundWorkerInitializeConnection(const char *dbname, const char *username, uint32 flags) +{ + BackgroundWorker *worker = t_thrd.bgworker_cxt.my_bgworker_entry; + + /* XXX is this the right errcode? */ + if (!(worker->bgw_flags & BGWORKER_BACKEND_DATABASE_CONNECTION)) { + ereport(FATAL, + (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), + errmsg("database connection requirement not indicated during registration"))); + } + + t_thrd.proc_cxt.PostInit->SetDatabaseAndUser(dbname, InvalidOid, username, InvalidOid); + t_thrd.proc_cxt.PostInit->InitBackendWorker(); + + /* it had better not gotten out of "init" mode yet */ + if (!IsInitProcessingMode()) { + ereport(ERROR, + (errmsg("invalid processing mode in background worker"))); + } + SetProcessingMode(NormalProcessing); +} + +/* + * Connect background worker to a database using OIDs. + */ +void BackgroundWorkerInitializeConnectionByOid(Oid dboid, Oid useroid, uint32 flags) +{ + BackgroundWorker *worker = t_thrd.bgworker_cxt.my_bgworker_entry; + + /* XXX is this the right errcode? */ + if (!(worker->bgw_flags & BGWORKER_BACKEND_DATABASE_CONNECTION)) { + ereport(FATAL, + (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), + errmsg("database connection requirement not indicated during registration"))); + } + + t_thrd.proc_cxt.PostInit->SetDatabaseAndUser(NULL, dboid, NULL, useroid); + t_thrd.proc_cxt.PostInit->InitBackendWorker(); + + /* it had better not gotten out of "init" mode yet */ + if (!IsInitProcessingMode()) { + ereport(ERROR, + (errmsg("invalid processing mode in background worker"))); + } + SetProcessingMode(NormalProcessing); +} + +/* + * Block/unblock signals in a background worker + */ +void BackgroundWorkerBlockSignals(void) +{ + (void)PG_SETMASK(&t_thrd.libpq_cxt.BlockSig); +} + +void BackgroundWorkerUnblockSignals(void) +{ + (void)PG_SETMASK(&t_thrd.libpq_cxt.UnBlockSig); +} + + diff --git a/src/gausskernel/process/postmaster/postmaster.cpp b/src/gausskernel/process/postmaster/postmaster.cpp index d01e8fcd5..8a0da6b55 100755 --- a/src/gausskernel/process/postmaster/postmaster.cpp +++ b/src/gausskernel/process/postmaster/postmaster.cpp @@ -110,6 +110,7 @@ #include "job/job_scheduler.h" #include "job/job_worker.h" #include "postmaster/autovacuum.h" +#include "postmaster/bgworker_internals.h" #include "postmaster/pagewriter.h" #include "postmaster/fork_process.h" #include "postmaster/pgarch.h" @@ -311,6 +312,7 @@ static void reaper(SIGNAL_ARGS); static void sigusr1_handler(SIGNAL_ARGS); static void dummy_handler(SIGNAL_ARGS); static void CleanupBackend(ThreadId pid, int exitstatus); +static bool CleanupBackgroundWorker(ThreadId pid, int exitstatus); static const char* GetProcName(ThreadId pid); static void LogChildExit(int lev, const char* procname, ThreadId pid, int exitstatus); static void PostmasterStateMachine(void); @@ -366,6 +368,8 @@ static void check_and_reset_ha_listen_port(void); static void* cJSON_internal_malloc(size_t size); static bool NeedHeartbeat(); static ServerMode GetHaShmemMode(void); +static bool assign_backendlist_entry(RegisteredBgWorker *rw); +static void maybe_start_bgworkers(void); bool PMstateIsRun(void); @@ -380,6 +384,7 @@ bool PMstateIsRun(void); #define BACKEND_TYPE_TEMPBACKEND \ 0x0010 /* temp thread processing cancel signal \ or stream connection */ + #define BACKEND_TYPE_ALL 0x001F /* OR of all the above */ static int CountChildren(int target); @@ -1019,6 +1024,7 @@ void SetShmemCxt(void) g_instance.shmem_cxt.MaxBackends = g_instance.shmem_cxt.MaxConnections + g_instance.attr.attr_sql.job_queue_processes + g_instance.attr.attr_storage.autovacuum_max_workers + + g_instance.attr.attr_storage.max_background_workers + AUXILIARY_BACKENDS + AV_LAUNCHER_PROCS; g_instance.shmem_cxt.MaxReserveBackendId = g_instance.attr.attr_sql.job_queue_processes + @@ -5464,6 +5470,14 @@ static void reaper(SIGNAL_ARGS) continue; } + /* Was it one of our background workers? */ + if (CleanupBackgroundWorker(pid, (int)exitstatus)) + { + /* have it be restarted */ + g_instance.bgworker_cxt.have_crashed_worker = true; + continue; + } + /* * Else do standard backend child cleanup. */ @@ -5566,6 +5580,101 @@ static const char* GetProcName(ThreadId pid) } } +/* + * Scan the bgworkers list and see if the given PID (which has just stopped + * or crashed) is in it. Handle its shutdown if so, and return true. If not a + * bgworker, return false. + * + * This is heavily based on CleanupBackend. One important difference is that + * we don't know yet that the dying process is a bgworker, so we must be silent + * until we're sure it is. + */ +static bool CleanupBackgroundWorker(ThreadId pid, + int exitstatus) /* child's exit status */ +{ + char namebuf[MAXPGPATH]; + slist_mutable_iter iter; + + slist_foreach_modify(iter, &BackgroundWorkerList) { + RegisteredBgWorker *rw; + + rw = slist_container(RegisteredBgWorker, rw_lnode, iter.cur); + + if (rw->rw_pid != pid) { + continue; + } + +#ifdef WIN32 + /* see CleanupBackend */ + if (exitstatus == ERROR_WAIT_NO_CHILDREN) { + exitstatus = 0; + } +#endif + + int rc = snprintf_s(namebuf, MAXPGPATH, MAXPGPATH - 1, _("background worker \"%s\""), rw->rw_worker.bgw_type); + securec_check_ss_c(rc, "\0", "\0"); + + if (!EXIT_STATUS_0(exitstatus)) { + /* Record timestamp, so we know when to restart the worker. */ + rw->rw_crashed_at = GetCurrentTimestamp(); + } else { + /* Zero exit status means terminate */ + rw->rw_crashed_at = 0; + rw->rw_terminate = true; + } + + /* + * Additionally, for shared-memory-connected workers, just like a + * backend, any exit status other than 0 or 1 is considered a crash + * and causes a system-wide restart. + */ + if ((rw->rw_worker.bgw_flags & BGWORKER_SHMEM_ACCESS) != 0) { + if (!EXIT_STATUS_0(exitstatus) && !EXIT_STATUS_1(exitstatus)) { + HandleChildCrash(pid, exitstatus, namebuf); + return true; + } + } + + /* + * We must release the postmaster child slot whether this worker is + * connected to shared memory or not, but we only treat it as a crash + * if it is in fact connected. + */ + if (!ReleasePostmasterChildSlot(rw->rw_child_slot) && + (rw->rw_worker.bgw_flags & BGWORKER_SHMEM_ACCESS) != 0) { + HandleChildCrash(pid, exitstatus, namebuf); + return true; + } + + /* Get it out of the BackendList and clear out remaining data */ + DLRemove(&rw->rw_backend->elem); + + /* + * It's possible that this background worker started some OTHER + * background worker and asked to be notified when that worker started + * or stopped. If so, cancel any notifications destined for the + * now-dead backend. + */ + if (rw->rw_backend->bgworker_notify) { + BackgroundWorkerStopNotifications(rw->rw_pid); + } + + BackendArrayRemove(rw->rw_backend); + + rw->rw_backend = NULL; + rw->rw_pid = 0; + rw->rw_child_slot = 0; + ReportBackgroundWorkerExit(&iter); /* report child death */ + + LogChildExit(EXIT_STATUS_0(exitstatus) ? DEBUG1 : LOG, + namebuf, pid, exitstatus); + + return true; + } + + return false; +} + /* * CleanupBackend -- cleanup after terminated backend. * @@ -5629,6 +5738,18 @@ static void CleanupBackend(ThreadId pid, int exitstatus) /* child's exit status. BackendArrayRemove(bp); } + if (bp->bgworker_notify) + { + /* + * This backend may have been slated to receive SIGUSR1 when + * some background worker started or stopped. Cancel those + * notifications, as we don't want to signal PIDs that are not + * PostgreSQL backends. This gets skipped in the (probably + * very common) case where the backend has never requested any + * such notifications. + */ + BackgroundWorkerStopNotifications(bp->pid); + } DLRemove(curr); break; } @@ -6881,6 +7002,16 @@ static void sigusr1_handler(SIGNAL_ARGS) gs_signal_setmask(&t_thrd.libpq_cxt.BlockSig, NULL); + /* Process background worker state change. */ + if (CheckPostmasterSignal(PMSIGNAL_BACKGROUND_WORKER_CHANGE)) + { + BackgroundWorkerStateChange(); + g_instance.bgworker_cxt.start_worker_needed = true; + } + if (g_instance.bgworker_cxt.start_worker_needed || g_instance.bgworker_cxt.have_crashed_worker) { + maybe_start_bgworkers(); + } + /* * RECOVERY_STARTED and BEGIN_HOT_STANDBY signals are ignored in * unexpected states. If the startup process quickly starts up, completes @@ -7699,6 +7830,300 @@ int MaxLivePostmasterChildren(void) return 6 * g_instance.shmem_cxt.MaxBackends; } +/* + * Start a new bgworker. + * Starting time conditions must have been checked already. + * + * Returns true on success, false on failure. + * In either case, update the RegisteredBgWorker's state appropriately. + * + * This code is heavily based on autovacuum.c, q.v. + */ +static bool do_start_bgworker(RegisteredBgWorker *rw) +{ + ThreadId worker_pid = InvalidPid; + + Assert(rw->rw_pid == 0); + + /* + * Allocate and assign the Backend element. Note we must do this before + * forking, so that we can handle failures (out of memory or child-process + * slots) cleanly. + * + * Treat failure as though the worker had crashed. That way, the + * postmaster will wait a bit before attempting to start it again; if we + * tried again right away, most likely we'd find ourselves hitting the + * same resource-exhaustion condition. + */ + if (!assign_backendlist_entry(rw)) { + rw->rw_crashed_at = GetCurrentTimestamp(); + return false; + } + + ereport(DEBUG1, + (errmsg("starting background worker process \"%s\"", + rw->rw_worker.bgw_name))); + + Backend* bn = rw->rw_backend; + void* bgWorkerShmAddr = GetBackgroundWorkerShmAddr(rw->rw_shmem_slot); + switch ((worker_pid = initialize_util_thread(BACKGROUND_WORKER, bgWorkerShmAddr))) { + case (ThreadId)-1: + /* in postmaster, fork failed ... */ + ereport(LOG, + (errmsg("could not fork worker process: %m"))); + /* undo what assign_backendlist_entry did */ + (void)ReleasePostmasterChildSlot(rw->rw_child_slot); + bn->pid = 0; + rw->rw_child_slot = 0; + rw->rw_backend = NULL; + /* mark entry as crashed, so we'll try again later */ + rw->rw_crashed_at = GetCurrentTimestamp(); + break; + + default: + /* in postmaster, fork successful ... */ + rw->rw_pid = worker_pid; + bn->pid = rw->rw_pid; + ReportBackgroundWorkerPID(rw); + /* add new worker to lists of backends */ + DLInitElem(&bn->elem, bn); + DLAddHead(g_instance.backend_list, &bn->elem); + + return true; + } + + return false; +} + +/* + * Does the current postmaster state require starting a worker with the + * specified start_time? + */ +static bool +bgworker_should_start_now(BgWorkerStartTime start_time) +{ + switch (pmState) { + case PM_NO_CHILDREN: + case PM_WAIT_DEAD_END: + case PM_SHUTDOWN_2: + case PM_SHUTDOWN: + case PM_WAIT_BACKENDS: + case PM_WAIT_READONLY: + case PM_WAIT_BACKUP: + break; + + case PM_RUN: + if (start_time == BgWorkerStart_RecoveryFinished) { + return true; + } + /* fall through */ + case PM_HOT_STANDBY: + if (start_time == BgWorkerStart_ConsistentState) { + return true; + } + /* fall through */ + case PM_RECOVERY: + case PM_STARTUP: + case PM_INIT: + if (start_time == BgWorkerStart_PostmasterStart) { + return true; + } + /* fall through */ + } + + return false; +} + +/* + * Allocate the Backend struct for a connected background worker, but don't + * add it to the list of backends just yet. + * + * On failure, return false without changing any worker state. + * + * Some info from the Backend is copied into the passed rw. + */ +static bool +assign_backendlist_entry(RegisteredBgWorker *rw) +{ + Backend* bn = NULL; + + /* + * Check that database state allows another connection. Currently the + * only possible failure is CAC_TOOMANY, so we just log an error message + * based on that rather than checking the error code precisely. + */ + if (canAcceptConnections(false) != CAC_OK) + { + ereport(LOG, + (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED), + errmsg("no slot available for new worker process"))); + return false; + } + + int slot = AssignPostmasterChildSlot(); + + bn = AssignFreeBackEnd(slot); + + if (bn == NULL) { + ereport(LOG, (errcode(ERRCODE_OUT_OF_MEMORY), errmsg("out of memory"))); + return false; + } + + /* + * Compute the cancel key that will be assigned to this session. We + * probably don't need cancel keys for background workers, but we'd better + * have something random in the field to prevent unfriendly people from + * sending cancels to them. + */ + GenerateCancelKey(false); + bn->cancel_key = t_thrd.proc_cxt.MyCancelKey; + bn->child_slot = t_thrd.proc_cxt.MyPMChildSlot = slot; + bn->is_autovacuum = false; + bn->dead_end = false; + bn->bgworker_notify = false; + rw->rw_backend = bn; + rw->rw_child_slot = bn->child_slot; + + return true; +} + +/* + * If the time is right, start background worker(s). + * + * As a side effect, the bgworker control variables are set or reset + * depending on whether more workers may need to be started. + * + * We limit the number of workers started per call, to avoid consuming the + * postmaster's attention for too long when many such requests are pending. + * As long as start_worker_needed is true, ServerLoop will not block and will + * call this function again after dealing with any other issues. + */ +static void maybe_start_bgworkers(void) +{ +#define MAX_BGWORKERS_TO_LAUNCH 100 + int num_launched = 0; + TimestampTz now = 0; + slist_mutable_iter iter; + + /* + * During crash recovery, we have no need to be called until the state + * transition out of recovery. + */ + if (g_instance.fatal_error) { + g_instance.bgworker_cxt.start_worker_needed = false; + g_instance.bgworker_cxt.have_crashed_worker = false; + return; + } + + /* Don't need to be called again unless we find a reason for it below */ + g_instance.bgworker_cxt.start_worker_needed = false; + g_instance.bgworker_cxt.have_crashed_worker = false; + + slist_foreach_modify(iter, &BackgroundWorkerList) { + RegisteredBgWorker *rw; + + rw = slist_container(RegisteredBgWorker, rw_lnode, iter.cur); + + /* ignore if already running */ + if (rw->rw_pid != 0) { + continue; + } + + /* if marked for death, clean up and remove from list */ + if (rw->rw_terminate) { + ForgetBackgroundWorker(&iter); + continue; + } + + /* + * If this worker has crashed previously, maybe it needs to be + * restarted (unless on registration it specified it doesn't want to + * be restarted at all). Check how long ago did a crash last happen. + * If the last crash is too recent, don't start it right away; let it + * be restarted once enough time has passed. + */ + if (rw->rw_crashed_at != 0) { + if (rw->rw_worker.bgw_restart_time == BGW_NEVER_RESTART) { + ThreadId notify_pid = rw->rw_worker.bgw_notify_pid; + + ForgetBackgroundWorker(&iter); + + /* Report worker is gone now. */ + if (notify_pid != 0) { + (void)gs_signal_send(notify_pid, SIGUSR1); + } + + continue; + } + + /* read system time only when needed */ + if (now == 0) { + now = GetCurrentTimestamp(); + } + + if (!TimestampDifferenceExceeds(rw->rw_crashed_at, now, + rw->rw_worker.bgw_restart_time * 1000)) { + /* Set flag to remember that we have workers to start later */ + g_instance.bgworker_cxt.have_crashed_worker = true; + continue; + } + } + + if (bgworker_should_start_now(rw->rw_worker.bgw_start_time)) { + /* reset crash time before trying to start worker */ + rw->rw_crashed_at = 0; + + /* + * Try to start the worker. + * + * On failure, give up processing workers for now, but set + * start_worker_needed so we'll come back here on the next iteration + * of ServerLoop to try again. (We don't want to wait, because + * there might be additional ready-to-run workers.) We could set + * have_crashed_worker as well, since this worker is now marked + * crashed, but there's no need because the next run of this + * function will do that. + */ + if (!do_start_bgworker(rw)) { + g_instance.bgworker_cxt.start_worker_needed = true; + return; + } + + /* + * If we've launched as many workers as allowed, quit, but have + * ServerLoop call us again to look for additional ready-to-run + * workers. There might not be any, but we'll find out the next + * time we run. + */ + if (++num_launched >= MAX_BGWORKERS_TO_LAUNCH) { + g_instance.bgworker_cxt.start_worker_needed = true; + return; + } + } + } +} + +/* + * When a backend asks to be notified about worker state changes, we + * set a flag in its backend entry. The background worker machinery needs + * to know when such backends exit. + */ +bool +PostmasterMarkPIDForWorkerNotify(ThreadId pid) +{ + int count = MaxLivePostmasterChildren(); + for (int i = 0; i < count; ++i) { + Backend* bp = &g_instance.backend_array[i]; + if (bp->pid == pid) + { + bp->bgworker_notify = true; + return true; + } + } + return false; +} + + #ifdef EXEC_BACKEND #ifndef WIN32 #define write_inheritable_socket(dest, src) ((*(dest) = (src))) @@ -7965,6 +8390,7 @@ Backend* AssignFreeBackEnd(int slot) bn->pid = 0; bn->cancel_key = 0; bn->dead_end = false; + bn->bgworker_notify = false; return bn; } @@ -9952,7 +10378,7 @@ int GaussDbThreadMain(knl_thread_arg* arg) commAuxiliaryMain(); proc_exit(0); } break; - + #ifdef ENABLE_MULTIPLE_NODES case COMM_POOLER_CLEAN: { InitProcessAndShareMemory(); @@ -9960,6 +10386,14 @@ int GaussDbThreadMain(knl_thread_arg* arg) proc_exit(0); } break; #endif + + case BACKGROUND_WORKER: { + IsBackgroundWorker = true; + InitProcessAndShareMemory(); + StartBackgroundWorker(arg->payload); + proc_exit(0); + } break; + default: ereport(PANIC, (errmsg("unsupport thread role type %d", arg->role))); break; @@ -10011,7 +10445,8 @@ static GaussdbThreadEntry GaussdbThreadEntryGate[] = {GaussDbThreadMain, GaussDbThreadMain, GaussDbThreadMain, GaussDbThreadMain, - GaussDbThreadMain}; + GaussDbThreadMain, + GaussDbThreadMain}; const char* GaussdbThreadName[] = {"main", "worker", @@ -10055,7 +10490,8 @@ const char* GaussdbThreadName[] = {"main", "communicator receiver flower", "communicator receiver loop", "communicator auxiliary", - "communicator pooler auto cleaner"}; + "communicator pooler auto cleaner", + "background worker"}; GaussdbThreadEntry GetThreadEntry(knl_thread_role role) { diff --git a/src/gausskernel/process/threadpool/knl_instance.cpp b/src/gausskernel/process/threadpool/knl_instance.cpp index 38794aac4..dad582020 100755 --- a/src/gausskernel/process/threadpool/knl_instance.cpp +++ b/src/gausskernel/process/threadpool/knl_instance.cpp @@ -290,6 +290,7 @@ static void knl_g_wlm_init(knl_g_wlm_context* wlm_cxt) static void knl_g_shmem_init(knl_g_shmem_context* shmem_cxt) { + shmem_cxt->max_parallel_workers = 8; shmem_cxt->MaxBackends = 100; shmem_cxt->MaxReserveBackendId = (AUXILIARY_BACKENDS + AV_LAUNCHER_PROCS); shmem_cxt->ThreadPoolGroupNum = 0; @@ -316,6 +317,12 @@ static void knl_g_numa_init(knl_g_numa_context* numa_cxt) numa_cxt->allocIndex = 0; } +static void knl_g_bgworker_init(knl_g_bgworker_context* bgworker_cxt) +{ + bgworker_cxt->start_worker_needed = true; + bgworker_cxt->have_crashed_worker = false; +} + void knl_instance_init() { g_instance.binaryupgrade = false; @@ -363,6 +370,7 @@ void knl_instance_init() knl_g_dw_init(&g_instance.dw_cxt); knl_g_xlog_init(&g_instance.xlog_cxt); knl_g_numa_init(&g_instance.numa_cxt); + knl_g_bgworker_init(&g_instance.bgworker_cxt); MemoryContextSwitchTo(old_cxt); diff --git a/src/gausskernel/process/threadpool/knl_thread.cpp b/src/gausskernel/process/threadpool/knl_thread.cpp index 3822f053b..1d0ca27a9 100755 --- a/src/gausskernel/process/threadpool/knl_thread.cpp +++ b/src/gausskernel/process/threadpool/knl_thread.cpp @@ -1405,6 +1405,12 @@ void knl_thread_mot_init() knl_t_mot_init(&t_thrd.mot_cxt); } +void knl_t_bgworker_init(knl_t_bgworker_context* bgworker_cxt) +{ + bgworker_cxt->background_worker_data = NULL; + bgworker_cxt->my_bgworker_entry = NULL; +} + void knl_thread_init(knl_thread_role role) { t_thrd.role = role; diff --git a/src/gausskernel/storage/ipc/ipci.cpp b/src/gausskernel/storage/ipc/ipci.cpp index 508a9abcb..fa8a0df83 100755 --- a/src/gausskernel/storage/ipc/ipci.cpp +++ b/src/gausskernel/storage/ipc/ipci.cpp @@ -37,6 +37,7 @@ #include "pgxc/nodemgr.h" #endif #include "postmaster/autovacuum.h" +#include "postmaster/bgworker_internals.h" #include "postmaster/bgwriter.h" #include "postmaster/postmaster.h" #include "replication/slot.h" @@ -136,6 +137,7 @@ void CreateSharedMemoryAndSemaphores(bool makePrivate, int port) size = add_size(size, CLOGShmemSize()); size = add_size(size, CSNLOGShmemSize()); size = add_size(size, TwoPhaseShmemSize()); + size = add_size(size, BackgroundWorkerShmemSize()); size = add_size(size, MultiXactShmemSize()); size = add_size(size, LWLockShmemSize()); size = add_size(size, ProcArrayShmemSize()); @@ -275,6 +277,7 @@ void CreateSharedMemoryAndSemaphores(bool makePrivate, int port) { TwoPhaseShmemInit(); } + BackgroundWorkerShmemInit(); /* * Set up shared-inval messaging diff --git a/src/gausskernel/storage/ipc/procsignal.cpp b/src/gausskernel/storage/ipc/procsignal.cpp index d7d92a989..3308fe14e 100644 --- a/src/gausskernel/storage/ipc/procsignal.cpp +++ b/src/gausskernel/storage/ipc/procsignal.cpp @@ -306,6 +306,7 @@ void procsignal_sigusr1_handler(SIGNAL_ARGS) if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN)) RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN); + SetLatch(&t_thrd.proc->procLatch); latch_sigusr1_handler(); errno = save_errno; diff --git a/src/gausskernel/storage/lmgr/lwlocknames.txt b/src/gausskernel/storage/lmgr/lwlocknames.txt index 602047f4d..1489b44f6 100644 --- a/src/gausskernel/storage/lmgr/lwlocknames.txt +++ b/src/gausskernel/storage/lmgr/lwlocknames.txt @@ -96,3 +96,4 @@ GPCCommitLock 88 GPCClearLock 89 GPCTimelineLock 90 TsTagsCacheLock 91 +BackgroundWorkerLock 92 \ No newline at end of file diff --git a/src/gausskernel/storage/lmgr/proc.cpp b/src/gausskernel/storage/lmgr/proc.cpp index ce2eb19b0..aba442a0c 100755 --- a/src/gausskernel/storage/lmgr/proc.cpp +++ b/src/gausskernel/storage/lmgr/proc.cpp @@ -212,7 +212,9 @@ static void FiniNuma(int code, Datum arg) * So, now we grab enough semaphores to support the desired max number * of backends immediately at initialization --- if the sysadmin has set * MaxConnections or autovacuum_max_workers higher than his kernel will - * support, he'll find out sooner rather than later. + * support, he'll find out sooner rather than later. (The number of + * background worker processes registered by loadable modules is also taken + * into consideration.) * * Another reason for creating semaphores here is that the semaphore * implementation typically requires us to create semaphores in the @@ -240,6 +242,7 @@ void InitProcGlobal(void) #endif g_instance.proc_base->freeProcs = NULL; g_instance.proc_base->autovacFreeProcs = NULL; + g_instance.proc_base->bgworkerFreeProcs = NULL; g_instance.proc_base->pgjobfreeProcs = NULL; g_instance.proc_base->startupProc = NULL; g_instance.proc_base->startupProcPid = 0; @@ -252,10 +255,10 @@ void InitProcGlobal(void) /* * Create and initialize all the PGPROC structures we'll need. There are - * four separate consumers: (1) normal backends, (2) autovacuum workers - * and the autovacuum launcher, (3) auxiliary processes, and (4) prepared - * transactions. Each PGPROC structure is dedicated to exactly one of - * these purposes, and they do not move between groups. + * five separate consumers: (1) normal backends, (2) autovacuum workers + * and the autovacuum launcher, (3) background workers, (4) auxiliary processes, + * and (5) prepared transactions. Each PGPROC structure is dedicated to exactly + * one of these purposes, and they do not move between groups. */ PGPROC *initProcs[MAX_NUMA_NODE] = {0}; @@ -331,7 +334,7 @@ void InitProcGlobal(void) procs[i]->nodeno = i % nNumaNodes; /* - * Newly created PGPROCs for normal backends or for autovacuum must be + * Newly created PGPROCs for normal backends, autovacuum and bgworkers must be * queued up on the appropriate free list. Because there can only * ever be a small, fixed number of auxiliary processes, no free list * is used in that case; InitAuxiliaryProcess() instead uses a linear @@ -347,13 +350,23 @@ void InitProcGlobal(void) /* PGPROC for pg_job backend, add to pgjobfreeProcs list, 1 for Job Schedule Lancher */ procs[i]->links.next = (SHM_QUEUE *)g_instance.proc_base->pgjobfreeProcs; g_instance.proc_base->pgjobfreeProcs = procs[i]; - } else if (i < g_instance.shmem_cxt.MaxBackends) { + } else if (i < g_instance.shmem_cxt.MaxConnections + AUXILIARY_BACKENDS + + g_instance.attr.attr_sql.job_queue_processes + 1 + + g_instance.attr.attr_storage.autovacuum_max_workers + + AV_LAUNCHER_PROCS) { /* * PGPROC for AV launcher/worker, add to autovacFreeProcs list - * list size is autovacuum_max_workers + AUTOVACUUM_LAUNCHERS + * list size is autovacuum_max_workers + AV_LAUNCHER_PROCS */ procs[i]->links.next = (SHM_QUEUE *)g_instance.proc_base->autovacFreeProcs; g_instance.proc_base->autovacFreeProcs = procs[i]; + } else if (i < g_instance.shmem_cxt.MaxBackends) { + /* + * PGPROC for bgworker, add to bgworkerFreeProcs list + * list size is max_background_workers + */ + procs[i]->links.next = (SHM_QUEUE *)g_instance.proc_base->bgworkerFreeProcs; + g_instance.proc_base->bgworkerFreeProcs = procs[i]; } /* Initialize myProcLocks[] shared memory queues. */ @@ -463,6 +476,8 @@ void InitProcess(void) t_thrd.proc = g_instance.proc_base->autovacFreeProcs; else if (IsJobSchedulerProcess() || IsJobWorkerProcess()) t_thrd.proc = g_instance.proc_base->pgjobfreeProcs; + else if (IsBackgroundWorker) + t_thrd.proc = g_instance.proc_base->bgworkerFreeProcs; else { #ifndef __USE_NUMA t_thrd.proc = g_instance.proc_base->freeProcs; @@ -478,6 +493,8 @@ void InitProcess(void) g_instance.proc_base->autovacFreeProcs = (PGPROC *)t_thrd.proc->links.next; else if (IsJobSchedulerProcess() || IsJobWorkerProcess()) g_instance.proc_base->pgjobfreeProcs = (PGPROC *)t_thrd.proc->links.next; + else if (IsBackgroundWorker) + g_instance.proc_base->bgworkerFreeProcs = (PGPROC *)t_thrd.proc->links.next; else { #ifndef __USE_NUMA g_instance.proc_base->freeProcs = (PGPROC *)t_thrd.proc->links.next; @@ -1036,6 +1053,11 @@ static void ProcKill(int code, Datum arg) } else if (IsJobSchedulerProcess() || IsJobWorkerProcess()) { t_thrd.proc->links.next = (SHM_QUEUE *)g_instance.proc_base->pgjobfreeProcs; g_instance.proc_base->pgjobfreeProcs = t_thrd.proc; + } + else if (IsBackgroundWorker) + { + t_thrd.proc->links.next = (SHM_QUEUE *)g_instance.proc_base->bgworkerFreeProcs; + g_instance.proc_base->bgworkerFreeProcs = t_thrd.proc; } else { t_thrd.proc->links.next = (SHM_QUEUE *)g_instance.proc_base->freeProcs; g_instance.proc_base->freeProcs = t_thrd.proc; diff --git a/src/include/gs_thread.h b/src/include/gs_thread.h index e32e21efe..dc0bfaf1b 100755 --- a/src/include/gs_thread.h +++ b/src/include/gs_thread.h @@ -97,6 +97,7 @@ typedef enum knl_thread_role { COMM_RECEIVER, COMM_AUXILIARY, COMM_POOLER_CLEAN, + BACKGROUND_WORKER, // should be last valid thread. THREAD_ENTRY_BOUND, diff --git a/src/include/knl/knl_guc/knl_instance_attr_storage.h b/src/include/knl/knl_guc/knl_instance_attr_storage.h index fc9805f80..ff905a4b6 100755 --- a/src/include/knl/knl_guc/knl_instance_attr_storage.h +++ b/src/include/knl/knl_guc/knl_instance_attr_storage.h @@ -68,6 +68,7 @@ typedef struct knl_instance_attr_storage { int max_replication_slots; int replication_type; int autovacuum_max_workers; + int max_background_workers; int64 autovacuum_freeze_max_age; int wal_level; /* User specified maximum number of recovery threads. */ diff --git a/src/include/knl/knl_instance.h b/src/include/knl/knl_instance.h index a7eba7c4c..edac6dfd4 100644 --- a/src/include/knl/knl_instance.h +++ b/src/include/knl/knl_instance.h @@ -485,6 +485,7 @@ typedef struct knl_g_libpq_context { typedef struct knl_g_shmem_context { int MaxConnections; + int max_parallel_workers; int MaxBackends; int MaxReserveBackendId; int ThreadPoolGroupNum; @@ -511,6 +512,12 @@ typedef struct knl_g_numa_context { size_t allocIndex; } knl_g_numa_context; +typedef struct knl_g_bgworker_context { + /* set when there's a worker that needs to be started up */ + volatile bool start_worker_needed; + volatile bool have_crashed_worker; +} knl_g_bgworker_context; + typedef struct knl_instance_context { knl_virtual_role role; volatile int status; @@ -582,6 +589,7 @@ typedef struct knl_instance_context { knl_g_rto_context rto_cxt; knl_g_xlog_context xlog_cxt; knl_g_numa_context numa_cxt; + knl_g_bgworker_context bgworker_cxt; } knl_instance_context; extern void knl_instance_init(); diff --git a/src/include/knl/knl_thread.h b/src/include/knl/knl_thread.h index 4a2939477..21a08dde1 100644 --- a/src/include/knl/knl_thread.h +++ b/src/include/knl/knl_thread.h @@ -69,6 +69,7 @@ #include "openssl/ossl_typ.h" #include "workload/qnode.h" #include "tcop/dest.h" +#include "postmaster/bgworker.h" #define MAX_PATH_LEN 1024 @@ -2702,6 +2703,11 @@ typedef struct knl_t_mot_context { unsigned int mbindFlags; } knl_t_mot_context; +typedef struct knl_t_bgworker_context { + BackgroundWorkerArray *background_worker_data; + BackgroundWorker *my_bgworker_entry; +} knl_t_bgworker_context; + /* thread context. */ typedef struct knl_thrd_context { knl_thread_role role; @@ -2799,6 +2805,7 @@ typedef struct knl_thrd_context { knl_t_heartbeat_context heartbeat_cxt; knl_t_poolcleaner_context poolcleaner_cxt; knl_t_mot_context mot_cxt; + knl_t_bgworker_context bgworker_cxt; } knl_thrd_context; extern void knl_thread_mot_init(); diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h index 3ca70ddc2..6d73c6960 100755 --- a/src/include/miscadmin.h +++ b/src/include/miscadmin.h @@ -28,6 +28,8 @@ #include "pgtime.h" /* for pg_time_t */ #include "libpq/libpq-be.h" +#define InvalidPid ((ThreadId)(-1)) + #define PG_BACKEND_VERSIONSTR "gaussdb " DEF_GS_VERSION "\n" /***************************************************************************** @@ -129,6 +131,7 @@ extern bool InplaceUpgradePrecommit; extern THR_LOCAL PGDLLIMPORT bool IsUnderPostmaster; extern THR_LOCAL PGDLLIMPORT char my_exec_path[]; +extern THR_LOCAL PGDLLIMPORT bool IsBackgroundWorker; #define MAX_QUERY_DOP (64) #define MIN_QUERY_DOP -(MAX_QUERY_DOP) @@ -232,7 +235,7 @@ extern bool InLocalUserIdChange(void); extern bool InSecurityRestrictedOperation(void); extern void GetUserIdAndContext(Oid* userid, bool* sec_def_context); extern void SetUserIdAndContext(Oid userid, bool sec_def_context); -extern void InitializeSessionUserId(const char* rolename); +extern void InitializeSessionUserId(const char* rolename, Oid role_id); extern void InitializeSessionUserIdStandalone(void); extern void SetSessionAuthorization(Oid userid, bool is_superuser); extern Oid GetCurrentRoleId(void); diff --git a/src/include/postmaster/bgworker.h b/src/include/postmaster/bgworker.h new file mode 100644 index 000000000..2c3d5c6e9 --- /dev/null +++ b/src/include/postmaster/bgworker.h @@ -0,0 +1,157 @@ +/* -------------------------------------------------------------------- + * bgworker.h + * POSTGRES pluggable background workers interface + * + * A background worker is a process able to run arbitrary, user-supplied code, + * including normal transactions. + * + * Any external module loaded via shared_preload_libraries can register a + * worker. Workers can also be registered dynamically at runtime. In either + * case, the worker process is forked from the postmaster and runs the + * user-supplied "main" function. This code may connect to a database and + * run transactions. Workers can remain active indefinitely, but will be + * terminated if a shutdown or crash occurs. + * + * If the fork() call fails in the postmaster, it will try again later. Note + * that the failure can only be transient (fork failure due to high load, + * memory pressure, too many processes, etc); more permanent problems, like + * failure to connect to a database, are detected later in the worker and dealt + * with just by having the worker exit normally. A worker which exits with + * a return code of 0 will never be restarted and will be removed from worker + * list. A worker which exits with a return code of 1 will be restarted after + * the configured restart interval (unless that interval is BGW_NEVER_RESTART). + * The TerminateBackgroundWorker() function can be used to terminate a + * dynamically registered background worker; the worker will be sent a SIGTERM + * and will not be restarted after it exits. Whenever the postmaster knows + * that a worker will not be restarted, it unregisters the worker, freeing up + * that worker's slot for use by a new worker. + * + * Note that there might be more than one worker in a database concurrently, + * and the same module may request more than one worker running the same (or + * different) code. + * + * + * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * IDENTIFICATION + * src/include/postmaster/bgworker.h + * -------------------------------------------------------------------- + */ +#include "gs_thread.h" +#ifndef BGWORKER_H +#define BGWORKER_H + +/*--------------------------------------------------------------------- + * External module API. + *--------------------------------------------------------------------- + */ + +/* + * Pass this flag to have your worker be able to connect to shared memory. + */ +#define BGWORKER_SHMEM_ACCESS 0x0001 + +/* + * This flag means the bgworker requires a database connection. The connection + * is not established automatically; the worker must establish it later. + * It requires that BGWORKER_SHMEM_ACCESS was passed too. + */ +#define BGWORKER_BACKEND_DATABASE_CONNECTION 0x0002 + +/* + * This class is used internally for parallel queries, to keep track of the + * number of active parallel workers and make sure we never launch more than + * max_parallel_workers parallel workers at the same time. Third party + * background workers should not use this class. + */ +#define BGWORKER_CLASS_PARALLEL 0x0010 + +/* add additional bgworker classes here */ + +typedef void (*bgworker_main_type) (Datum main_arg); + +/* + * Points in time at which a bgworker can request to be started + */ +typedef enum { + BgWorkerStart_PostmasterStart, + BgWorkerStart_ConsistentState, + BgWorkerStart_RecoveryFinished +} BgWorkerStartTime; + +#define BGW_DEFAULT_RESTART_INTERVAL 60 +#define BGW_NEVER_RESTART -1 +#define BGW_MAXLEN 96 +#define BGW_EXTRALEN 128 + +typedef struct BackgroundWorker { + char bgw_name[BGW_MAXLEN]; + char bgw_type[BGW_MAXLEN]; + int bgw_flags; + BgWorkerStartTime bgw_start_time; + int bgw_restart_time; /* in seconds, or BGW_NEVER_RESTART */ + char bgw_library_name[BGW_MAXLEN]; + char bgw_function_name[BGW_MAXLEN]; + Datum bgw_main_arg; + char bgw_extra[BGW_EXTRALEN]; + ThreadId bgw_notify_pid; /* SIGUSR1 this backend on start/stop */ +} BackgroundWorker; + +typedef enum BgwHandleStatus { + BGWH_STARTED, /* worker is running */ + BGWH_NOT_YET_STARTED, /* worker hasn't been started yet */ + BGWH_STOPPED, /* worker has exited */ + BGWH_POSTMASTER_DIED /* postmaster died; worker status unclear */ +} BgwHandleStatus; + +struct BackgroundWorkerHandle; +typedef struct BackgroundWorkerHandle BackgroundWorkerHandle; +struct BackgroundWorkerArray; + +/* Register a new bgworker during shared_preload_libraries */ +extern void RegisterBackgroundWorker(BackgroundWorker *worker); + +/* Register a new bgworker from a regular backend */ +extern bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, + BackgroundWorkerHandle **handle); + +/* Query the status of a bgworker */ +extern BgwHandleStatus GetBackgroundWorkerPid(const BackgroundWorkerHandle *handle, + ThreadId *pidp); +extern BgwHandleStatus WaitForBackgroundWorkerStartup(const BackgroundWorkerHandle *handle, ThreadId *pid); +extern BgwHandleStatus WaitForBackgroundWorkerShutdown(const BackgroundWorkerHandle *handle); +extern const char *GetBackgroundWorkerTypeByPid(ThreadId pid); + +/* Terminate a bgworker */ +extern void TerminateBackgroundWorker(const BackgroundWorkerHandle *handle); + +/* + * Connect to the specified database, as the specified user. Only a worker + * that passed BGWORKER_BACKEND_DATABASE_CONNECTION during registration may + * call this. + * + * If username is NULL, bootstrapping superuser is used. + * If dbname is NULL, connection is made to no specific database; + * only shared catalogs can be accessed. + */ +extern void BackgroundWorkerInitializeConnection(const char *dbname, const char *username, uint32 flags = 1); + +/* Just like the above, but specifying database and user by OID. */ +extern void BackgroundWorkerInitializeConnectionByOid(Oid dboid, Oid useroid, uint32 flags = 1); + +/* + * Flags to BackgroundWorkerInitializeConnection et al + * + * + * Allow bypassing datallowconn restrictions when connecting to database + */ +#define BGWORKER_BYPASS_ALLOWCONN 1 + + +/* Block/unblock signals in a background worker process */ +extern void BackgroundWorkerBlockSignals(void); +extern void BackgroundWorkerUnblockSignals(void); + +#endif /* BGWORKER_H */ + diff --git a/src/include/postmaster/bgworker_internals.h b/src/include/postmaster/bgworker_internals.h new file mode 100644 index 000000000..fa57bdc4d --- /dev/null +++ b/src/include/postmaster/bgworker_internals.h @@ -0,0 +1,64 @@ +/* -------------------------------------------------------------------- + * bgworker_internals.h + * POSTGRES pluggable background workers internals + * + * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * IDENTIFICATION + * src/include/postmaster/bgworker_internals.h + * -------------------------------------------------------------------- + */ +#ifndef BGWORKER_INTERNALS_H +#define BGWORKER_INTERNALS_H + +#include "datatype/timestamp.h" +#include "lib/ilist.h" +#include "postmaster/bgworker.h" + +/* + * Maximum possible value of parallel workers. + */ +#define MAX_PARALLEL_WORKER_LIMIT 1024 + +struct BackgroundWorkerSlot; + +/* + * List of background workers, private to postmaster. + * + * A worker that requests a database connection during registration will have + * rw_backend set, and will be present in BackendList. Note: do not rely on + * rw_backend being non-NULL for shmem-connected workers! + */ +typedef struct RegisteredBgWorker { + BackgroundWorker rw_worker; /* its registry entry */ + struct Backend *rw_backend; /* its BackendList entry, or NULL */ + ThreadId rw_pid; /* 0 if not running */ + int rw_child_slot; + TimestampTz rw_crashed_at; /* if not 0, time it last crashed */ + int rw_shmem_slot; + bool rw_terminate; + slist_node rw_lnode; /* list link */ +} RegisteredBgWorker; + +extern THR_LOCAL slist_head BackgroundWorkerList; + +extern Size BackgroundWorkerShmemSize(void); +extern void BackgroundWorkerShmemInit(void); +extern void BackgroundWorkerStateChange(void); +extern void ForgetBackgroundWorker(slist_mutable_iter *cur); +extern void ReportBackgroundWorkerPID(const RegisteredBgWorker *); +extern void ReportBackgroundWorkerExit(slist_mutable_iter *cur); +extern void BackgroundWorkerStopNotifications(ThreadId pid); +extern void ResetBackgroundWorkerCrashTimes(void); + +/* Function to start a background worker, called from postmaster.c */ +extern void StartBackgroundWorker(void* bgWorkerSlotShmAddr) ; + +#ifdef EXEC_BACKEND +extern void* GetBackgroundWorkerShmAddr(int slotno); +extern BackgroundWorker *BackgroundWorkerEntry(const BackgroundWorkerSlot* bgWorkerSlotShmAddr); +#endif + +#endif /* BGWORKER_INTERNALS_H */ + diff --git a/src/include/postmaster/postmaster.h b/src/include/postmaster/postmaster.h index 6c2ec7c0f..5070a7996 100644 --- a/src/include/postmaster/postmaster.h +++ b/src/include/postmaster/postmaster.h @@ -249,4 +249,5 @@ extern void GenerateCancelKey(bool isThreadPoolSession); extern bool SignalCancelAllBackEnd(); extern bool IsLocalAddr(Port* port); extern uint64_t mc_timers_us(void); +extern bool PostmasterMarkPIDForWorkerNotify(ThreadId); #endif /* _POSTMASTER_H */ diff --git a/src/include/storage/pmsignal.h b/src/include/storage/pmsignal.h index 07e213fac..e78bebb20 100755 --- a/src/include/storage/pmsignal.h +++ b/src/include/storage/pmsignal.h @@ -43,6 +43,7 @@ typedef enum { PMSIGNAL_ROLLBACK_STANDBY_PROMOTE, /* roll back standby promoting */ PMSIGNAL_START_PAGE_WRITER, /* start a new page writer thread */ PMSIGNAL_START_THREADPOOL_WORKER, /* start thread pool woker */ + PMSIGNAL_BACKGROUND_WORKER_CHANGE, /* background worker state change */ NUM_PMSIGNALS /* Must be last value of enum! */ } PMSignalReason; diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index b0a14ae3e..e76d80d86 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -304,6 +304,8 @@ typedef struct PROC_HDR { PGPROC* freeProcs; /* Head of list of autovacuum's free PGPROC structures */ PGPROC* autovacFreeProcs; + /* Head of list of bgworker free PGPROC structures */ + PGPROC* bgworkerFreeProcs; /* Head of list of pg_job's free PGPROC structures */ PGPROC* pgjobfreeProcs; /* First pgproc waiting for group XID clear */ diff --git a/src/include/threadpool/threadpool_worker.h b/src/include/threadpool/threadpool_worker.h index 7ec2cf647..2c22e5ce0 100755 --- a/src/include/threadpool/threadpool_worker.h +++ b/src/include/threadpool/threadpool_worker.h @@ -54,6 +54,7 @@ typedef struct Backend { bool is_autovacuum; /* is it an autovacuum process? */ volatile bool dead_end; /* is it going to send an quit? */ volatile int flag; + bool bgworker_notify; /* gets bgworker start/stop notifications */ Dlelem elem; /* list link in BackendList */ } Backend; diff --git a/src/include/utils/postinit.h b/src/include/utils/postinit.h index 2525a8fec..dccc83079 100755 --- a/src/include/utils/postinit.h +++ b/src/include/utils/postinit.h @@ -54,7 +54,7 @@ public: ~PostgresInitializer(); - void SetDatabaseAndUser(const char* in_dbname, Oid dboid, const char* username); + void SetDatabaseAndUser(const char* in_dbname, Oid dboid, const char* username, Oid useroid = InvalidOid); void GetDatabaseName(char* out_dbname); @@ -91,6 +91,8 @@ public: const char* m_username; + Oid m_useroid; + private: void InitThread();