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

move global variable to instance or thread

This commit is contained in:
jiang_jianyu 2020-08-31 21:01:43 +08:00
parent b3e7623efe
commit 26a06c95b8
15 changed files with 103 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/common/backend/libpq/pqcomm.cpp
View File

@ -154,7 +154,7 @@ static int socket_putmessage_noblock(char msgtype, const char *s, size_t len);
static void socket_startcopyout(void);
static void socket_endcopyout(bool errorAbort);
static PQcommMethods PqCommSocketMethods = {
static const PQcommMethods PqCommSocketMethods = {
socket_comm_reset,
socket_flush,
socket_flush_if_writable,
@ -165,7 +165,10 @@ static PQcommMethods PqCommSocketMethods = {
socket_endcopyout
};
THR_LOCAL PQcommMethods *PqCommMethods = &PqCommSocketMethods;
void PqCommMethods_init()
{
t_thrd.msqueue_cxt.PqCommMethods = &PqCommSocketMethods;
}
extern bool FencedUDFMasterMode;

63
src/common/backend/libpq/pqmq.cpp
View File

@ -21,12 +21,6 @@
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
static THR_LOCAL shm_mq *pq_mq;
static THR_LOCAL shm_mq_handle *pq_mq_handle;
static THR_LOCAL bool pq_mq_busy = false;
static THR_LOCAL ThreadId pq_mq_parallel_master_pid = 0;
static THR_LOCAL BackendId pq_mq_parallel_master_backend_id = InvalidBackendId;
static void mq_comm_reset(void);
static int mq_flush(void);
static int mq_flush_if_writable(void);
@ -36,7 +30,7 @@ static int mq_putmessage_noblock(char msgtype, const char *s, size_t len);
static void mq_startcopyout(void);
static void mq_endcopyout(bool errorAbort);
static THR_LOCAL PQcommMethods PqCommMqMethods = {
static const PQcommMethods PqCommMqMethods = {
mq_comm_reset,
mq_flush,
mq_flush_if_writable,
@ -47,32 +41,29 @@ static THR_LOCAL PQcommMethods PqCommMqMethods = {
mq_endcopyout
};
static THR_LOCAL PQcommMethods *save_PqCommMethods;
static THR_LOCAL CommandDest save_whereToSendOutput;
static THR_LOCAL ProtocolVersion save_FrontendProtocol;
/*
* Arrange to redirect frontend/backend protocol messages to a message queue.
*/
void pq_redirect_to_shm_mq(shm_mq_handle *mqh)
{
save_PqCommMethods = PqCommMethods;
save_whereToSendOutput = CommandDest(t_thrd.postgres_cxt.whereToSendOutput);
save_FrontendProtocol = FrontendProtocol;
t_thrd.msqueue_cxt.save_PqCommMethods = t_thrd.msqueue_cxt.PqCommMethods;
t_thrd.msqueue_cxt.save_whereToSendOutput = CommandDest(t_thrd.postgres_cxt.whereToSendOutput);
t_thrd.msqueue_cxt.save_FrontendProtocol = FrontendProtocol;
PqCommMethods = &PqCommMqMethods;
pq_mq_handle = mqh;
t_thrd.msqueue_cxt.PqCommMethods = &PqCommMqMethods;
t_thrd.msqueue_cxt.pq_mq = shm_mq_get_queue(mqh);
t_thrd.msqueue_cxt.pq_mq_handle = mqh;
t_thrd.postgres_cxt.whereToSendOutput = static_cast<int>(DestRemote);
FrontendProtocol = PG_PROTOCOL_LATEST;
}
void pq_stop_redirect_to_shm_mq(void)
{
PqCommMethods = save_PqCommMethods;
t_thrd.postgres_cxt.whereToSendOutput = static_cast<int>(save_whereToSendOutput);
FrontendProtocol = save_FrontendProtocol;
pq_mq = NULL;
pq_mq_handle = NULL;
t_thrd.msqueue_cxt.PqCommMethods = t_thrd.msqueue_cxt.save_PqCommMethods;
t_thrd.postgres_cxt.whereToSendOutput = static_cast<int>(t_thrd.msqueue_cxt.save_whereToSendOutput);
FrontendProtocol = t_thrd.msqueue_cxt.save_FrontendProtocol;
t_thrd.msqueue_cxt.pq_mq = NULL;
t_thrd.msqueue_cxt.pq_mq_handle = NULL;
}
/*
@ -81,9 +72,9 @@ void pq_stop_redirect_to_shm_mq(void)
*/
void pq_set_parallel_master(ThreadId pid, BackendId backend_id)
{
Assert(PqCommMethods == &PqCommMqMethods);
pq_mq_parallel_master_pid = pid;
pq_mq_parallel_master_backend_id = backend_id;
Assert(t_thrd.msqueue_cxt.PqCommMethods == &PqCommMqMethods);
t_thrd.msqueue_cxt.pq_mq_parallel_master_pid = pid;
t_thrd.msqueue_cxt.pq_mq_parallel_master_backend_id = backend_id;
}
static void mq_comm_reset(void)
@ -127,10 +118,10 @@ static int mq_putmessage(char msgtype, const char *s, size_t len)
* queueing the message would amount to indefinitely postponing the
* response to the interrupt. So we do this instead.
*/
if (pq_mq_busy) {
if (pq_mq_handle != NULL)
shm_mq_detach(pq_mq_handle);
pq_mq_handle = NULL;
if (t_thrd.msqueue_cxt.pq_mq_busy) {
if (t_thrd.msqueue_cxt.pq_mq_handle != NULL)
shm_mq_detach(t_thrd.msqueue_cxt.pq_mq_handle);
t_thrd.msqueue_cxt.pq_mq_handle = NULL;
return EOF;
}
@ -140,24 +131,24 @@ static int mq_putmessage(char msgtype, const char *s, size_t len)
* be generated late in the shutdown sequence, after all DSMs have already
* been detached.
*/
if (pq_mq_handle == NULL)
if (t_thrd.msqueue_cxt.pq_mq_handle == NULL)
return 0;
pq_mq_busy = true;
t_thrd.msqueue_cxt.pq_mq_busy = true;
iov[0].data = &msgtype;
iov[0].len = 1;
iov[1].data = s;
iov[1].len = len;
Assert(pq_mq_handle != NULL);
Assert(t_thrd.msqueue_cxt.pq_mq_handle != NULL);
for (;;) {
result = shm_mq_sendv(pq_mq_handle, iov, 2, true);
result = shm_mq_sendv(t_thrd.msqueue_cxt.pq_mq_handle, iov, 2, true);
if (pq_mq_parallel_master_pid != 0)
(void)SendProcSignal(pq_mq_parallel_master_pid,PROCSIG_PARALLEL_MESSAGE,
pq_mq_parallel_master_backend_id);
if (t_thrd.msqueue_cxt.pq_mq_parallel_master_pid != 0)
(void)SendProcSignal(t_thrd.msqueue_cxt.pq_mq_parallel_master_pid,PROCSIG_PARALLEL_MESSAGE,
t_thrd.msqueue_cxt.pq_mq_parallel_master_backend_id);
if (result != SHM_MQ_WOULD_BLOCK)
break;
@ -167,7 +158,7 @@ static int mq_putmessage(char msgtype, const char *s, size_t len)
CHECK_FOR_INTERRUPTS();
}
pq_mq_busy = false;
t_thrd.msqueue_cxt.pq_mq_busy = false;
Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
if (result != SHM_MQ_SUCCESS)

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

@ -51,7 +51,6 @@ 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;

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

@ -846,7 +846,7 @@ void InitializeSessionUserIdStandalone(void)
*/
AssertState(!IsUnderPostmaster || IsAutoVacuumWorkerProcess() ||
IsJobSchedulerProcess() || IsJobWorkerProcess() || AM_WAL_SENDER ||
IsBackgroundWorker);
t_thrd.bgworker_cxt.is_background_worker);
/* In pooler stateless reuse mode, to reset session userid */
if (!g_instance.attr.attr_network.PoolerStatelessReuse) {

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

@ -1495,7 +1495,7 @@ void PostgresInitializer::InitSession()
if (!IsUnderPostmaster) {
CheckAtLeastOneRoles();
SetSuperUserStandalone();
} else if (IsBackgroundWorker) {
} else if (t_thrd.bgworker_cxt.is_background_worker) {
if (m_username == NULL && !OidIsValid(m_useroid)) {
InitializeSessionUserIdStandalone();
m_isSuperUser = true;

19
src/gausskernel/process/postmaster/bgworker.cpp
View File

@ -33,11 +33,6 @@
#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.
@ -159,7 +154,7 @@ void BackgroundWorkerShmemInit(void)
* correspondence between the postmaster's private list and the array
* in shared memory.
*/
slist_foreach(siter, &BackgroundWorkerList) {
slist_foreach(siter, &t_thrd.bgworker_cxt.background_worker_list) {
BackgroundWorkerSlot *slot = &t_thrd.bgworker_cxt.background_worker_data->slot[slotno];
RegisteredBgWorker *rw;
@ -198,7 +193,7 @@ static RegisteredBgWorker * FindRegisteredWorkerBySlotNumber(int slotno)
{
slist_iter siter;
slist_foreach(siter, &BackgroundWorkerList) {
slist_foreach(siter, &t_thrd.bgworker_cxt.background_worker_list) {
RegisteredBgWorker *rw = slist_container(RegisteredBgWorker, rw_lnode, siter.cur);
if (rw->rw_shmem_slot == slotno) {
return rw;
@ -371,7 +366,7 @@ void BackgroundWorkerStateChange(void)
(errmsg("registering background worker \"%s\"",
rw->rw_worker.bgw_name)));
slist_push_head(&BackgroundWorkerList, &rw->rw_lnode);
slist_push_head(&t_thrd.bgworker_cxt.background_worker_list, &rw->rw_lnode);
}
}
@ -475,7 +470,7 @@ void BackgroundWorkerStopNotifications(ThreadId pid)
{
slist_iter siter;
slist_foreach(siter, &BackgroundWorkerList)
slist_foreach(siter, &t_thrd.bgworker_cxt.background_worker_list)
{
RegisteredBgWorker *rw = slist_container(RegisteredBgWorker, rw_lnode, siter.cur);
if (rw->rw_worker.bgw_notify_pid == pid) {
@ -495,7 +490,7 @@ void ResetBackgroundWorkerCrashTimes(void)
{
slist_mutable_iter iter;
slist_foreach_modify(iter, &BackgroundWorkerList)
slist_foreach_modify(iter, &t_thrd.bgworker_cxt.background_worker_list)
{
RegisteredBgWorker *rw = slist_container(RegisteredBgWorker, rw_lnode, iter.cur);
@ -705,7 +700,7 @@ void StartBackgroundWorker(void* bgWorkerSlotShmAddr)
(errmsg("unable to find bgworker entry")));
}
IsBackgroundWorker = true;
t_thrd.bgworker_cxt.is_background_worker = true;
/* Identify myself via ps */
init_ps_display(worker->bgw_name, "", "", "");
@ -889,7 +884,7 @@ void RegisterBackgroundWorker(BackgroundWorker *worker)
rw->rw_crashed_at = 0;
rw->rw_terminate = false;
slist_push_head(&BackgroundWorkerList, &rw->rw_lnode);
slist_push_head(&t_thrd.bgworker_cxt.background_worker_list, &rw->rw_lnode);
}
/*

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

@ -5595,7 +5595,7 @@ static bool CleanupBackgroundWorker(ThreadId pid,
char namebuf[MAXPGPATH];
slist_mutable_iter iter;
slist_foreach_modify(iter, &BackgroundWorkerList) {
slist_foreach_modify(iter, &t_thrd.bgworker_cxt.background_worker_list) {
RegisteredBgWorker *rw;
rw = slist_container(RegisteredBgWorker, rw_lnode, iter.cur);
@ -8019,7 +8019,7 @@ static void maybe_start_bgworkers(void)
g_instance.bgworker_cxt.start_worker_needed = false;
g_instance.bgworker_cxt.have_crashed_worker = false;
slist_foreach_modify(iter, &BackgroundWorkerList) {
slist_foreach_modify(iter, &t_thrd.bgworker_cxt.background_worker_list) {
RegisteredBgWorker *rw;
rw = slist_container(RegisteredBgWorker, rw_lnode, iter.cur);
@ -10388,7 +10388,7 @@ int GaussDbThreadMain(knl_thread_arg* arg)
#endif
case BACKGROUND_WORKER: {
IsBackgroundWorker = true;
t_thrd.bgworker_cxt.is_background_worker = true;
InitProcessAndShareMemory();
StartBackgroundWorker(arg->payload);
proc_exit(0);

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

@ -67,6 +67,7 @@
#include "utils/postinit.h"
#include "utils/relmapper.h"
#include "workload/workload.h"
#include "libpq/pqcomm.h"
THR_LOCAL knl_thrd_context t_thrd;
@ -1417,6 +1418,22 @@ void knl_t_bgworker_init(knl_t_bgworker_context* bgworker_cxt)
{
bgworker_cxt->background_worker_data = NULL;
bgworker_cxt->my_bgworker_entry = NULL;
bgworker_cxt->is_background_worker = false;
bgworker_cxt->background_worker_list = SLIST_STATIC_INIT(background_worker_list);
}
void knl_t_msqueue_init(knl_t_msqueue_context* msqueue_cxt)
{
msqueue_cxt->pq_mq = NULL;
msqueue_cxt->pq_mq_handle = NULL;
msqueue_cxt->pq_mq_busy = false;
msqueue_cxt->pq_mq_parallel_master_pid = 0;
msqueue_cxt->pq_mq_parallel_master_backend_id = InvalidBackendId;
msqueue_cxt->save_PqCommMethods = NULL;
msqueue_cxt->save_whereToSendOutput = DestDebug;
msqueue_cxt->save_FrontendProtocol = PG_PROTOCOL_LATEST;
//msqueue_cxt->PqCommMethods = NULL;
PqCommMethods_init();
}
void knl_thread_init(knl_thread_role role)
@ -1507,6 +1524,8 @@ void knl_thread_init(knl_thread_role role)
knl_t_poolcleaner_init(&t_thrd.poolcleaner_cxt);
knl_t_mot_init(&t_thrd.mot_cxt);
knl_t_autonomous_init(&t_thrd.autonomous_cxt);
knl_t_bgworker_init(&t_thrd.bgworker_cxt);
knl_t_msqueue_init(&t_thrd.msqueue_cxt);
}
void knl_thread_set_name(const char* name)

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

@ -476,7 +476,7 @@ 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)
else if (t_thrd.bgworker_cxt.is_background_worker)
t_thrd.proc = g_instance.proc_base->bgworkerFreeProcs;
else {
#ifndef __USE_NUMA
@ -493,7 +493,7 @@ 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)
else if (t_thrd.bgworker_cxt.is_background_worker)
g_instance.proc_base->bgworkerFreeProcs = (PGPROC *)t_thrd.proc->links.next;
else {
#ifndef __USE_NUMA
@ -1054,7 +1054,7 @@ static void ProcKill(int code, Datum arg)
t_thrd.proc->links.next = (SHM_QUEUE *)g_instance.proc_base->pgjobfreeProcs;
g_instance.proc_base->pgjobfreeProcs = t_thrd.proc;
}
else if (IsBackgroundWorker)
else if (t_thrd.bgworker_cxt.is_background_worker)
{
t_thrd.proc->links.next = (SHM_QUEUE *)g_instance.proc_base->bgworkerFreeProcs;
g_instance.proc_base->bgworkerFreeProcs = t_thrd.proc;

4
src/include/gtm/pqcomm.h
View File

@ -29,6 +29,10 @@ typedef struct {
size_t salen;
} SockAddr;
void PqCommMethods_init();
typedef unsigned int uint32; /* == 32 bits */
typedef uint32 ProtocolVersion; /* FE/BE protocol version number */
/*
* In protocol 3.0 and later, the startup packet length is not fixed, but
* we set an arbitrary limit on it anyway. This is just to prevent simple

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

@ -2716,8 +2716,28 @@ typedef struct knl_t_mot_context {
typedef struct knl_t_bgworker_context {
BackgroundWorkerArray *background_worker_data;
BackgroundWorker *my_bgworker_entry;
bool is_background_worker;
/*
* The postmaster's list of registered background workers, in private memory.
*/
slist_head background_worker_list;
} knl_t_bgworker_context;
struct shm_mq;
struct shm_mq_handle;
struct PQcommMethods;
typedef struct knl_t_msqueue_context {
shm_mq *pq_mq;
shm_mq_handle *pq_mq_handle;
bool pq_mq_busy;
ThreadId pq_mq_parallel_master_pid;
BackendId pq_mq_parallel_master_backend_id;
const PQcommMethods *save_PqCommMethods;
CommandDest save_whereToSendOutput;
ProtocolVersion save_FrontendProtocol;
const PQcommMethods *PqCommMethods;
} knl_t_msqueue_context;
/* thread context. */
typedef struct knl_thrd_context {
knl_thread_role role;
@ -2817,6 +2837,7 @@ typedef struct knl_thrd_context {
knl_t_poolcleaner_context poolcleaner_cxt;
knl_t_mot_context mot_cxt;
knl_t_bgworker_context bgworker_cxt;
knl_t_msqueue_context msqueue_cxt;
} knl_thrd_context;
extern void knl_thread_mot_init();

22
src/include/libpq/libpq.h
View File

@ -35,7 +35,7 @@ typedef struct {
} u;
} PQArgBlock;
typedef struct {
struct PQcommMethods{
void (*comm_reset) (void);
int (*flush) (void);
int (*flush_if_writable) (void);
@ -44,20 +44,18 @@ typedef struct {
int (*putmessage_noblock) (char msgtype, const char* s, size_t len);
void (*startcopyout) (void);
void (*endcopyout) (bool errorAbort);
} PQcommMethods;
};
extern PGDLLIMPORT THR_LOCAL PQcommMethods *PqCommMethods;
#define pq_comm_reset() (PqCommMethods->comm_reset())
#define pq_flush() (PqCommMethods->flush())
#define pq_flush_if_writable() (PqCommMethods->flush_if_writable())
#define pq_is_send_pending() (PqCommMethods->is_send_pending())
#define pq_comm_reset() (t_thrd.msqueue_cxt.PqCommMethods->comm_reset())
#define pq_flush() (t_thrd.msqueue_cxt.PqCommMethods->flush())
#define pq_flush_if_writable() (t_thrd.msqueue_cxt.PqCommMethods->flush_if_writable())
#define pq_is_send_pending() (t_thrd.msqueue_cxt.PqCommMethods->is_send_pending())
#define pq_putmessage(msgtype, s, len) \
(PqCommMethods->putmessage(msgtype, s, len))
(t_thrd.msqueue_cxt.PqCommMethods->putmessage(msgtype, s, len))
#define pq_putmessage_noblock(msgtype, s, len) \
(PqCommMethods->putmessage_noblock(msgtype, s, len))
#define pq_startcopyout() (PqCommMethods->startcopyout())
#define pq_endcopyout(errorAbort) (PqCommMethods->endcopyout(errorAbort))
(t_thrd.msqueue_cxt.PqCommMethods->putmessage_noblock(msgtype, s, len))
#define pq_startcopyout() (t_thrd.msqueue_cxt.PqCommMethods->startcopyout())
#define pq_endcopyout(errorAbort) (t_thrd.msqueue_cxt.PqCommMethods->endcopyout(errorAbort))
/*
* External functions.

2
src/include/libpq/pqcomm.h
View File

@ -239,6 +239,8 @@ typedef struct ConnPack {
ConnectStreamPacket cp;
} ConnPack;
void PqCommMethods_init();
/*
* A client can also start by sending stop query request
*/

1
src/include/miscadmin.h
View File

@ -131,7 +131,6 @@ 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)

2
src/include/postmaster/bgworker_internals.h
View File

@ -41,8 +41,6 @@ typedef struct RegisteredBgWorker {
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);