bfwl2021
/
openGauss-server
forked from huawei/openGauss-server
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

support background workers from PG

This commit is contained in:
jiang_jianyu 2020-08-25 21:17:24 +08:00
parent dc967cf640
commit b3c0ecfb4c
27 changed files with 2129 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/common/backend/port/unix_latch.cpp
View File

@ -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 */

1
src/common/backend/utils/init/globals.cpp
View File

@ -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;

43
src/common/backend/utils/init/miscinit.cpp
View File

@ -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) {

24
src/common/backend/utils/init/postinit.cpp
View File

@ -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;
}

34
src/common/backend/utils/misc/guc.cpp
View File

@ -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;

2
src/gausskernel/optimizer/commands/async.cpp
View File

@ -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)
*

2
src/gausskernel/process/postmaster/Makefile
View File

@ -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

1309
src/gausskernel/process/postmaster/bgworker.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

442
src/gausskernel/process/postmaster/postmaster.cpp
View File

@ -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<MASTER>,
GaussDbThreadMain<COMM_RECEIVERFLOWER>,
GaussDbThreadMain<COMM_RECEIVER>,
GaussDbThreadMain<COMM_AUXILIARY>,
GaussDbThreadMain<COMM_POOLER_CLEAN>};
GaussDbThreadMain<COMM_POOLER_CLEAN>,
GaussDbThreadMain<BACKGROUND_WORKER>};
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)
{

8
src/gausskernel/process/threadpool/knl_instance.cpp
View File

@ -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);

6
src/gausskernel/process/threadpool/knl_thread.cpp
View File

@ -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;

3
src/gausskernel/storage/ipc/ipci.cpp
View File

@ -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

1
src/gausskernel/storage/ipc/procsignal.cpp
View File

@ -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;

1
src/gausskernel/storage/lmgr/lwlocknames.txt
View File

@ -96,3 +96,4 @@ GPCCommitLock 88
GPCClearLock 89
GPCTimelineLock 90
TsTagsCacheLock 91
BackgroundWorkerLock 92

38
src/gausskernel/storage/lmgr/proc.cpp
View File

@ -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;

1
src/include/gs_thread.h
View File

@ -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,

1
src/include/knl/knl_guc/knl_instance_attr_storage.h
View File

@ -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. */

8
src/include/knl/knl_instance.h
View File

@ -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();

7
src/include/knl/knl_thread.h
View File

@ -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();

5
src/include/miscadmin.h
View File

@ -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);

157
src/include/postmaster/bgworker.h Normal file
View File

@ -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 */

64
src/include/postmaster/bgworker_internals.h Normal file
View File

@ -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 */

1
src/include/postmaster/postmaster.h
View File

@ -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 */

1
src/include/storage/pmsignal.h
View File

@ -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;

2
src/include/storage/proc.h
View File

@ -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 */

1
src/include/threadpool/threadpool_worker.h
View File

@ -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;

4
src/include/utils/postinit.h
View File

@ -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();