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

!1033 fix gpc bug and optimize gpc performance 

Merge pull request !1033 from gentle_hu/2.0.0
This commit is contained in:
opengauss-bot 2021-07-27 12:41:51 +00:00 committed by Gitee
parent a189abd40e c6b1e6dd61
commit c5661c88a9
15 changed files with 246 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/common/backend/utils/cache/inval.cpp vendored
View File

@ -101,6 +101,7 @@
#include "catalog/pg_proc.h"
#include "commands/prepare.h"
#include "miscadmin.h"
#include "postmaster/postmaster.h"
#include "storage/sinval.h"
#include "storage/smgr.h"
#include "utils/inval.h"
@ -561,7 +562,11 @@ void LocalExecuteInvalidationMessage(SharedInvalidationMessage* msg)
if (ENABLE_GPC) {
bool check = GlobalPlanCache::MsgCheck(msg);
#ifdef ENABLE_MULTIPLE_NODES
if (check == true && u_sess->pcache_cxt.gpc_remote_msg == false) {
#else
if (check == true && (u_sess->pcache_cxt.gpc_remote_msg == false || pmState == PM_HOT_STANDBY)) {
#endif
u_sess->pcache_cxt.gpc_in_ddl = true;
}
}

56
src/common/backend/utils/cache/plancache.cpp vendored
View File

@ -103,8 +103,6 @@ static bool ChooseCustomPlan(CachedPlanSource* plansource, ParamListInfo boundPa
static bool IsForceCustomplan(CachedPlanSource *plansource);
static bool IsDeleteLimit(CachedPlanSource* plansource, ParamListInfo boundParams);
static double cached_plan_cost(CachedPlan* plan);
static void AcquireExecutorLocks(List* stmt_list, bool acquire);
static void AcquirePlannerLocks(List* stmt_list, bool acquire);
static void ScanQueryForLocks(Query* parsetree, bool acquire);
static bool ScanQueryWalker(Node* node, bool* acquire);
static TupleDesc PlanCacheComputeResultDesc(List* stmt_list);
@ -722,10 +720,9 @@ List* RevalidateCachedQuery(CachedPlanSource* plansource, bool has_lp)
Assert(plansource->is_valid);
return NIL;
}
/* if is shared plan, we should acquire plan lock for this transaction */
/* if is shared plan, we should acquire plan lock before check recreate plan */
if (plansource->gpc.status.InShareTable()) {
Assert(plansource->is_valid);
AcquirePlannerLocks(plansource->query_list, true);
return NIL;
}
/*
@ -1959,18 +1956,20 @@ CachedPlanSource* CopyCachedPlan(CachedPlanSource* plansource, bool is_share)
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot copy a one-shot cached plan")));
if (ENABLE_GPC && is_share == true) {
source_context = AllocSetContextCreate(GLOBAL_PLANCACHE_MEMCONTEXT,
"GPCCachedPlanSource",
ALLOCSET_SMALL_MINSIZE,
ALLOCSET_SMALL_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE,
SHARED_CONTEXT);
source_context = AllocSetContextCreate(GLOBAL_PLANCACHE_MEMCONTEXT,
"GPCCachedPlanSource",
ALLOCSET_SMALL_MINSIZE,
ALLOCSET_SMALL_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE,
SHARED_CONTEXT);
ResourceOwnerEnlargeGMemContext(t_thrd.utils_cxt.TopTransactionResourceOwner);
ResourceOwnerRememberGMemContext(t_thrd.utils_cxt.TopTransactionResourceOwner, source_context);
} else {
source_context = AllocSetContextCreate(u_sess->cache_mem_cxt,
"CachedPlanSource",
ALLOCSET_SMALL_MINSIZE,
ALLOCSET_SMALL_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
source_context = AllocSetContextCreate(u_sess->cache_mem_cxt,
"CachedPlanSource",
ALLOCSET_SMALL_MINSIZE,
ALLOCSET_SMALL_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
}
@ -2004,19 +2003,19 @@ CachedPlanSource* CopyCachedPlan(CachedPlanSource* plansource, bool is_share)
newsource->context = source_context;
if (ENABLE_GPC && is_share == true) {
querytree_context = AllocSetContextCreate(source_context,
"GPCCachedPlanQuery",
ALLOCSET_SMALL_MINSIZE,
ALLOCSET_SMALL_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE,
SHARED_CONTEXT);
querytree_context = AllocSetContextCreate(source_context,
"GPCCachedPlanQuery",
ALLOCSET_SMALL_MINSIZE,
ALLOCSET_SMALL_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE,
SHARED_CONTEXT);
}
else {
querytree_context = AllocSetContextCreate(source_context,
"CachedPlanQuery",
ALLOCSET_SMALL_MINSIZE,
ALLOCSET_SMALL_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
querytree_context = AllocSetContextCreate(source_context,
"CachedPlanQuery",
ALLOCSET_SMALL_MINSIZE,
ALLOCSET_SMALL_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
}
MemoryContextSwitchTo(querytree_context);
newsource->query_list = (List*)copyObject(plansource->query_list);
@ -2042,6 +2041,7 @@ CachedPlanSource* CopyCachedPlan(CachedPlanSource* plansource, bool is_share)
newsource->opFusionObj = NULL;
newsource->is_checked_opfusion = false;
newsource->spi_signature = plansource->spi_signature;
newsource->gplan_is_fqs = plansource->gplan_is_fqs;
#ifdef ENABLE_MOT
newsource->storageEngineType = SE_TYPE_UNSPECIFIED;
@ -2113,7 +2113,7 @@ List* CachedPlanGetTargetList(CachedPlanSource* plansource)
* AcquireExecutorLocks: acquire locks needed for execution of a cached plan;
* or release them if acquire is false.
*/
static void AcquireExecutorLocks(List* stmt_list, bool acquire)
void AcquireExecutorLocks(List* stmt_list, bool acquire)
{
ListCell* lc1 = NULL;
@ -2179,7 +2179,7 @@ static void AcquireExecutorLocks(List* stmt_list, bool acquire)
* fail if one has been dropped entirely --- we'll just transiently acquire
* a non-conflicting lock.
*/
static void AcquirePlannerLocks(List* stmt_list, bool acquire)
void AcquirePlannerLocks(List* stmt_list, bool acquire)
{
ListCell* lc = NULL;

9
src/gausskernel/optimizer/commands/prepare.cpp
View File

@ -236,12 +236,12 @@ void ExecuteQuery(ExecuteStmt* stmt, IntoClause* intoClause, const char* querySt
OpFusion *opFusionObj = (OpFusion *)(psrc->opFusionObj);
if (opFusionObj->IsGlobal()) {
opFusionObj = (OpFusion *)OpFusion::FusionFactory(opFusionObj->m_global->m_type,
u_sess->cache_mem_cxt, psrc, NULL, params);
u_sess->cache_mem_cxt, psrc, NULL, paramLI);
Assert(opFusionObj != NULL);
}
opFusionObj->setPreparedDestReceiver(dest);
opFusionObj->useOuterParameter(paramLI);
opFusionObj->setCurrentOpFusionObj((OpFusion*)psrc->opFusionObj);
opFusionObj->setCurrentOpFusionObj(opFusionObj);
CachedPlanSource* cps = opFusionObj->m_global->m_psrc;
bool needBucketId = cps != NULL && cps->gplan;
@ -720,10 +720,11 @@ void StorePreparedStatement(const char* stmt_name, CachedPlanSource* plansource,
static void FetchPreparedStatementCNGPC(PreparedStatement* entry, const char* stmt_name)
{
Assert (entry->plansource->magic == CACHEDPLANSOURCE_MAGIC);
bool hasGetLock = false;
/* check if need recreate */
if (g_instance.plan_cache->CheckRecreateCachePlan(entry->plansource)) {
if (g_instance.plan_cache->CheckRecreateCachePlan(entry->plansource, &hasGetLock)) {
entry->has_prepare_dn_stmt = false;
g_instance.plan_cache->RecreateCachePlan(entry->plansource, entry->stmt_name, entry, NULL, NULL);
g_instance.plan_cache->RecreateCachePlan(entry->plansource, entry->stmt_name, entry, NULL, NULL, hasGetLock);
}
#ifdef ENABLE_MULTIPLE_NODES
Assert (entry->plansource->lightProxyObj == NULL);

140
src/gausskernel/process/globalplancache/globalplancache.cpp
View File

@ -72,6 +72,9 @@ CompareSearchPath(struct OverrideSearchPath* path1, struct OverrideSearchPath* p
if (list_difference_oid(path1->schemas, path2->schemas) != NULL) {
return false;
}
if (list_difference_oid(path2->schemas, path1->schemas) != NULL) {
return false;
}
return true;
}
/*
@ -349,10 +352,15 @@ CachedPlanSource* GlobalPlanCache::Fetch(const char *query_string, uint32 query_
LWLockRelease(GetMainLWLockByIndex(lock_id));
return NULL;
} else {
entry->val.plansource->gpc.status.AddRefcount();
CachedPlanSource* psrc = entry->val.plansource;
if (!psrc->gpc.status.IsValid()) {
LWLockRelease(GetMainLWLockByIndex(lock_id));
MoveIntoInvalidPlanList(psrc);
return NULL;
}
psrc->gpc.status.AddRefcount();
if (ENABLE_DN_GPC)
u_sess->pcache_cxt.private_refcount++;
CachedPlanSource* psrc = entry->val.plansource;
pg_atomic_fetch_add_u32(&entry->val.used_count, 1);
MemoryContextSwitchTo(oldcontext);
LWLockRelease(GetMainLWLockByIndex(lock_id));
@ -480,7 +488,7 @@ void GlobalPlanCache::RemoveEntry(uint32 htblIdx, GPCEntry *entry)
}
bool GlobalPlanCache::CheckRecreateCachePlan(CachedPlanSource* psrc)
bool GlobalPlanCache::CheckRecreateCachePlan(CachedPlanSource* psrc, bool* hasGetLock)
{
/*
* Start up a transaction command so we can run parse analysis etc. (Note
@ -489,6 +497,14 @@ bool GlobalPlanCache::CheckRecreateCachePlan(CachedPlanSource* psrc)
*/
start_xact_command();
Assert(psrc->magic == CACHEDPLANSOURCE_MAGIC);
/* get lock before check plan is valid or not, release it if need recreate plan */
if (psrc->gpc.status.InShareTable()) {
AcquirePlannerLocks(psrc->query_list, true);
if (psrc->gplan) {
AcquireExecutorLocks(psrc->gplan->stmt_list, true);
}
*hasGetLock = true;
}
#ifdef ENABLE_MULTIPLE_NODES
if (IS_PGXC_COORDINATOR && !psrc->gpc.status.InShareTable()) {
@ -526,7 +542,14 @@ bool GlobalPlanCache::CheckRecreateSPICachePlan(SPIPlanPtr spi_plan)
Assert(spi_plan->magic == _SPI_PLAN_MAGIC);
foreach(cell, spi_plan->plancache_list) {
CachedPlanSource* plansource = (CachedPlanSource*)lfirst(cell);
if (CheckRecreateCachePlan(plansource)) {
bool hasGetLock = false;
if (CheckRecreateCachePlan(plansource, &hasGetLock)) {
if (hasGetLock) {
AcquirePlannerLocks(plansource->query_list, false);
if (plansource->gplan) {
AcquireExecutorLocks(plansource->gplan->stmt_list, false);
}
}
return true;
}
}
@ -548,51 +571,93 @@ void GlobalPlanCache::RecreateSPICachePlan(SPIPlanPtr spiplan)
if (!oldsource->gpc.status.InShareTable())
continue;
GPC_LOG("recreate spi cachedplan", oldsource, 0);
RecreateCachePlan(oldsource, NULL, NULL, spiplan, cell);
RecreateCachePlan(oldsource, NULL, NULL, spiplan, cell, false);
}
/* pop error context stack */
t_thrd.log_cxt.error_context_stack = spi_err_context.previous;
Assert(SPIPlanCacheTableLookup(u_sess->SPI_cxt._current->spi_hash_key));
}
void GlobalPlanCache::RecreateCachePlan(CachedPlanSource* oldsource, const char* stmt_name,
PreparedStatement *entry, SPIPlanPtr spiplan, ListCell* spiplanCell)
void GlobalPlanCache::MoveIntoInvalidPlanList(CachedPlanSource* psrc)
{
if (psrc->gpc.status.InShareTable() && !psrc->gpc.status.IsValid()) {
GPCKey* key = psrc->gpc.key;
uint32 hashCode = GPCHashFunc((const void *) key, sizeof(*key));
uint32 bucket_id = GetBucket(hashCode);
int lock_id = m_array[bucket_id].lockId;
(void)LWLockAcquire(GetMainLWLockByIndex(lock_id), LW_EXCLUSIVE);
if (psrc->gpc.status.InShareTableInvalidList() == false) {
bool found = false;
hash_search(m_array[bucket_id].hash_tbl, (void *)key, HASH_REMOVE, &found);
if (unlikely(found == false))
elog(PANIC, "should found plan in gpc");
m_array[bucket_id].count--;
AddInvalidList(psrc);
}
LWLockRelease(GetMainLWLockByIndex(lock_id));
}
}
void GlobalPlanCache::RecreateCachePlan(CachedPlanSource* oldsource, const char* stmt_name, PreparedStatement *entry,
SPIPlanPtr spiplan, ListCell* spiplanCell, bool hasGetLock)
{
GPC_LOG("recreate plan", oldsource, oldsource->stmt_name);
CachedPlanSource *newsource = CopyCachedPlan(oldsource, true);
MemoryContext oldcxt = MemoryContextSwitchTo(newsource->context);
newsource->stream_enabled = IsStreamSupport();
u_sess->exec_cxt.CurrentOpFusionObj = NULL;
Assert (oldsource->gpc.status.IsSharePlan());
newsource->gpc.status.ShareInit();
// If the planSource is set to invalid, the AST must be analyzed again
// because the meta has changed.
newsource->is_valid = false;
bool has_lp = false;
if (spiplan != NULL) {
t_thrd.log_cxt.error_context_stack->arg = (void *)newsource->query_string;
newsource->spi_signature = oldsource->spi_signature;
newsource->parserSetup = spiplan->parserSetup;
newsource->parserSetupArg = spiplan->parserSetupArg;
} else if (IS_PGXC_DATANODE) {
newsource->stmt_name = pstrdup(stmt_name);
} else {
newsource->stmt_name = pstrdup(stmt_name);
#ifdef ENABLE_MULTIPLE_NODES
has_lp = (oldsource->single_exec_node != NULL && oldsource->gplan == NULL && oldsource->cplan == NULL);
/* clean session's datanode statment on cn */
if (has_lp) {
/* no lp in newsource, delete old lp */
GPCDropLPIfNecessary(stmt_name, false, true, NULL);
} else if (oldsource->gplan != NULL) {
/* Close any active planned Datanode statements, recreate in BuildCachedPlan later */
GPCCleanDatanodeStatement(oldsource->gplan->dn_stmt_num, stmt_name);
/* these operator may throw error, make sure shared plan is invalid first */
oldsource->gpc.status.SetStatus(GPC_INVALID);
oldsource->is_valid = false;
CachedPlanSource *newsource = NULL;
PG_TRY();
{
if (hasGetLock) {
AcquirePlannerLocks(oldsource->query_list, false);
if (oldsource->gplan) {
AcquireExecutorLocks(oldsource->gplan->stmt_list, false);
}
}
newsource = CopyCachedPlan(oldsource, true);
MemoryContext oldcxt = MemoryContextSwitchTo(newsource->context);
newsource->stream_enabled = IsStreamSupport();
u_sess->exec_cxt.CurrentOpFusionObj = NULL;
Assert (oldsource->gpc.status.IsSharePlan());
newsource->gpc.status.ShareInit();
// If the planSource is set to invalid, the AST must be analyzed again
// because the meta has changed.
newsource->is_valid = false;
bool has_lp = false;
if (spiplan != NULL) {
t_thrd.log_cxt.error_context_stack->arg = (void *)newsource->query_string;
newsource->spi_signature = oldsource->spi_signature;
newsource->parserSetup = spiplan->parserSetup;
newsource->parserSetupArg = spiplan->parserSetupArg;
} else if (IS_PGXC_DATANODE) {
newsource->stmt_name = pstrdup(stmt_name);
} else {
newsource->stmt_name = pstrdup(stmt_name);
#ifdef ENABLE_MULTIPLE_NODES
has_lp = (oldsource->single_exec_node != NULL && oldsource->gplan == NULL && oldsource->cplan == NULL);
/* clean session's datanode statment on cn */
if (has_lp) {
/* no lp in newsource, delete old lp */
GPCDropLPIfNecessary(stmt_name, false, true, NULL);
} else if (oldsource->gplan != NULL) {
/* Close any active planned Datanode statements, recreate in BuildCachedPlan later */
GPCCleanDatanodeStatement(oldsource->gplan->dn_stmt_num, stmt_name);
}
#endif
}
(void)RevalidateCachedQuery(newsource, has_lp);
MemoryContextSwitchTo(oldcxt);
}
(void)RevalidateCachedQuery(newsource, has_lp);
PG_CATCH();
{
/* catch only move invalid plansource into gpc invalid list when error occurs */
MoveIntoInvalidPlanList(oldsource);
PG_RE_THROW();
}
PG_END_TRY();
/* newsource has reference on session, forget resource owner */
ResourceOwnerForgetGMemContext(t_thrd.utils_cxt.TopTransactionResourceOwner, newsource->context);
newsource->next_saved = u_sess->pcache_cxt.first_saved_plan;
u_sess->pcache_cxt.first_saved_plan = newsource;
newsource->is_saved = true;
@ -610,7 +675,6 @@ void GlobalPlanCache::RecreateCachePlan(CachedPlanSource* oldsource, const char*
#endif
}
MemoryContextSwitchTo(oldcxt);
RemovePlanSource<ACTION_RECREATE>(oldsource, stmt_name);
}

25
src/gausskernel/process/globalplancache/globalplancache_view.cpp
View File

@ -96,28 +96,9 @@ GlobalPlanCache::GetStatus(uint32 *num)
stat_array[index].refcount = ps->gpc.status.GetRefCount();
stat_array[index].valid = ps->gpc.status.IsValid();
stat_array[index].DatabaseID = entry->key.env.plainenv.database_id;
StringInfoData all_schema_name;
ListCell *cell = NULL;
bool has_schema = false;
foreach(cell, entry->key.env.search_path->schemas) {
char* nspname = get_namespace_name(lfirst_oid(cell));
if (nspname != NULL) {
if (has_schema)
appendStringInfo(&all_schema_name, ", ");
else
initStringInfo(&all_schema_name);
appendStringInfo(&all_schema_name, "%s", nspname);
has_schema = true;
}
pfree_ext(nspname);
}
if (!has_schema) {
stat_array[index].schema_name = (char *)palloc0(sizeof(char) * NAMEDATALEN);
rc = memcpy_s(stat_array[index].schema_name, NAMEDATALEN, entry->key.env.schema_name, NAMEDATALEN);
securec_check(rc, "\0", "\0");
} else {
stat_array[index].schema_name = all_schema_name.data;
}
stat_array[index].schema_name = (char *)palloc0(sizeof(char) * NAMEDATALEN);
rc = memcpy_s(stat_array[index].schema_name, NAMEDATALEN, entry->key.env.schema_name, NAMEDATALEN);
securec_check(rc, "\0", "\0");
stat_array[index].params_num = ps->num_params;
stat_array[index].func_id = entry->key.spi_signature.func_oid;
bool printPlan = u_sess->attr.attr_sql.Debug_print_plan && entry->val.plansource->gplan &&

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

@ -4044,6 +4044,7 @@ static void SIGHUP_handler(SIGNAL_ARGS)
(void)SignalChildren(SIGHUP);
if (ENABLE_THREAD_POOL) {
g_threadPoolControler->GetSessionCtrl()->SigHupHandler();
g_threadPoolControler->GetScheduler()->SigHupHandler();
}
if (g_instance.pid_cxt.StartupPID != 0)

107
src/gausskernel/process/tcop/postgres.cpp
View File

@ -3169,6 +3169,40 @@ static void exec_parse_message(const char* query_string, /* string to execute */
if (ENABLE_DN_GPC)
CleanSessGPCPtr(u_sess);
is_named = (stmt_name[0] != '\0');
if (ENABLE_GPC) {
CachedPlanSource * plansource = g_instance.plan_cache->Fetch(query_string, strlen(query_string),
numParams, NULL);
if (plansource != NULL) {
bool hasGetLock = false;
if (is_named) {
if (ENABLE_CN_GPC)
StorePreparedStatementCNGPC(stmt_name, plansource, false, true);
else {
u_sess->pcache_cxt.cur_stmt_psrc = plansource;
if (g_instance.plan_cache->CheckRecreateCachePlan(plansource, &hasGetLock))
g_instance.plan_cache->RecreateCachePlan(plansource, stmt_name, NULL, NULL, NULL, hasGetLock);
}
goto pass_parsing;
} else {
if (ENABLE_DN_GPC)
u_sess->pcache_cxt.private_refcount--;
if (!g_instance.plan_cache->CheckRecreateCachePlan(plansource, &hasGetLock)) {
drop_unnamed_stmt();
u_sess->pcache_cxt.unnamed_stmt_psrc = plansource;
goto pass_parsing;
} else {
plansource->gpc.status.SubRefCount();
if (hasGetLock) {
AcquirePlannerLocks(plansource->query_list, false);
if (plansource->gplan) {
AcquireExecutorLocks(plansource->gplan->stmt_list, false);
}
}
}
}
}
}
/*
* Switch to appropriate context for constructing parsetrees.
*
@ -3183,24 +3217,7 @@ static void exec_parse_message(const char* query_string, /* string to execute */
* So in this case, we create the plancache entry's query_context here,
* and do all the parsing work therein.
*/
is_named = (stmt_name[0] != '\0');
if (is_named) {
if (ENABLE_GPC) {
CachedPlanSource * plansource = g_instance.plan_cache->Fetch(query_string, strlen(query_string),
numParams, NULL);
if (plansource != NULL) {
if (ENABLE_CN_GPC)
StorePreparedStatementCNGPC(stmt_name, plansource, false, true);
else {
u_sess->pcache_cxt.cur_stmt_psrc = plansource;
if (g_instance.plan_cache->CheckRecreateCachePlan(plansource))
g_instance.plan_cache->RecreateCachePlan(plansource, stmt_name, NULL, NULL, NULL);
}
goto pass_parsing;
}
}
/* Named prepared statement --- parse in u_sess->parser_cxt.temp_parse_message_context */
oldcontext = MemoryContextSwitchTo(u_sess->temp_mem_cxt);
} else {
@ -4196,6 +4213,7 @@ static void exec_bind_message(StringInfo input_message)
/* save the cursor in case of error */
int msg_cursor = input_message->cursor;
int nodeIdx = getSingleNodeIdx(input_message, psrc, stmt_name);
bool enable_unamed_gpc = psrc->gpc.status.InShareTable() && stmt_name[0] == '\0';
if (nodeIdx != -1) {
lightProxy* scn = NULL;
bool enable_gpc = (ENABLE_CN_GPC && stmt_name[0] != '\0');
@ -4212,7 +4230,12 @@ static void exec_bind_message(StringInfo input_message)
if (enable_gpc) {
scn->storeLpByStmtName(stmt_name);
psrc->lightProxyObj = NULL;
} else if (enable_unamed_gpc) {
Assert(u_sess->pcache_cxt.unnamed_gpc_lp == NULL);
Assert(psrc->lightProxyObj == NULL);
u_sess->pcache_cxt.unnamed_gpc_lp = scn;
} else {
Assert(!enable_unamed_gpc);
psrc->lightProxyObj = scn;
}
} else {
@ -5401,12 +5424,28 @@ static void drop_unnamed_stmt(void)
if (u_sess->pcache_cxt.unnamed_stmt_psrc) {
CachedPlanSource* psrc = u_sess->pcache_cxt.unnamed_stmt_psrc;
u_sess->pcache_cxt.unnamed_stmt_psrc = NULL;
if (ENABLE_GPC) {
GPC_LOG("drop private plansource", psrc, psrc->stmt_name);
}
DropCachedPlan(psrc);
if (ENABLE_GPC) {
GPC_LOG("drop private plansource end", psrc, 0);
if (psrc->gpc.status.InShareTable()) {
/* drop unnamed lightproxy */
if (u_sess->pcache_cxt.unnamed_gpc_lp) {
lightProxy* lp = u_sess->pcache_cxt.unnamed_gpc_lp;
u_sess->pcache_cxt.unnamed_gpc_lp = NULL;
Assert(lp->m_cplan == psrc);
Assert(lp->m_stmtName == NULL || lp->m_stmtName[0] == '\0');
if (lp->m_portalName == NULL || lightProxy::locateLightProxy(lp->m_portalName) == NULL) {
lightProxy::tearDown(lp);
}
}
GPC_LOG("delete shared unnamed plansource", psrc, psrc->stmt_name);
psrc->gpc.status.SubRefCount();
} else {
Assert(u_sess->pcache_cxt.unnamed_gpc_lp == NULL);
if (ENABLE_GPC) {
GPC_LOG("drop private plansource", psrc, psrc->stmt_name);
}
DropCachedPlan(psrc);
if (ENABLE_GPC) {
GPC_LOG("drop private plansource end", psrc, 0);
}
}
}
}
@ -7393,6 +7432,7 @@ int PostgresMain(int argc, char* argv[], const char* dbname, const char* usernam
}
}
u_sess->pcache_cxt.gpc_in_batch = false;
u_sess->pcache_cxt.gpc_in_try_store = false;
OpFusion::tearDown(u_sess->exec_cxt.CurrentOpFusionObj);
@ -8436,6 +8476,7 @@ int PostgresMain(int argc, char* argv[], const char* dbname, const char* usernam
OpFusion* curr = OpFusion::locateFusion(closeTarget);
if (curr != NULL) {
if (curr->IsGlobal()) {
curr->clean();
OpFusion::tearDown(curr);
} else {
curr->clean();
@ -9816,10 +9857,11 @@ static void exec_one_in_batch(CachedPlanSource* psrc, ParamListInfo params, int
DestReceiver* receiver = NULL;
bool completed = false;
bool hasGetLock = false;
if (ENABLE_GPC && psrc->stmt_name && psrc->stmt_name[0] != '\0' &&
g_instance.plan_cache->CheckRecreateCachePlan(psrc)) {
g_instance.plan_cache->RecreateCachePlan(psrc, stmt_name, pstmt, NULL, NULL);
g_instance.plan_cache->CheckRecreateCachePlan(psrc, &hasGetLock)) {
g_instance.plan_cache->RecreateCachePlan(psrc, stmt_name, pstmt, NULL, NULL, hasGetLock);
#ifdef ENABLE_MULTIPLE_NODES
psrc = IS_PGXC_DATANODE ? u_sess->pcache_cxt.cur_stmt_psrc : pstmt->plansource;
#else
@ -10358,6 +10400,8 @@ static void exec_batch_bind_execute(StringInfo input_message)
/* E message */
const char* exec_portal_name = NULL;
int max_rows;
/* reset gpc batch flag */
u_sess->pcache_cxt.gpc_in_batch = false;
/*
* Only support normal perf mode for PBE, as DestRemoteExecute can not send T message automatically.
@ -10578,7 +10622,7 @@ static void exec_batch_bind_execute(StringInfo input_message)
}
/* Make sure the querytree list is valid and we have parse-time locks */
if (!ENABLE_CN_GPC && psrc->single_exec_node != NULL)
if (psrc->single_exec_node != NULL)
RevalidateCachedQuery(psrc);
/* record the params set position for light cn to contruct batch message */
@ -10900,6 +10944,7 @@ static void exec_batch_bind_execute(StringInfo input_message)
/* 3.run for each dn */
lightProxy *scn = NULL;
bool enable_gpc = (ENABLE_CN_GPC && stmt_name[0] != '\0');
bool enable_unamed_gpc = psrc->gpc.status.InShareTable() && stmt_name[0] == '\0';
if (enable_gpc)
scn = lightProxy::locateLpByStmtName(stmt_name);
else
@ -10913,9 +10958,15 @@ static void exec_batch_bind_execute(StringInfo input_message)
if (enable_gpc) {
scn->storeLpByStmtName(stmt_name);
psrc->lightProxyObj = NULL;
} else if (enable_unamed_gpc) {
Assert(u_sess->pcache_cxt.unnamed_gpc_lp == NULL);
Assert(psrc->lightProxyObj == NULL);
u_sess->pcache_cxt.unnamed_gpc_lp = scn;
} else {
psrc->lightProxyObj = scn;
}
} else {
Assert(!enable_unamed_gpc);
}
if (enable_gpc) {
/* cngpc need fill gpc msg just like BuildCachedPlan. */
@ -10943,6 +10994,7 @@ static void exec_batch_bind_execute(StringInfo input_message)
if (ENABLE_GPC && stmt_name[0] != '\0') {
#ifdef ENABLE_MULTIPLE_NODES
psrc = IS_PGXC_DATANODE ? u_sess->pcache_cxt.cur_stmt_psrc : pstmt->plansource;
u_sess->pcache_cxt.gpc_in_batch = true;
#else
psrc = pstmt->plansource;
#endif
@ -10961,6 +11013,7 @@ static void exec_batch_bind_execute(StringInfo input_message)
if (ENABLE_GPC && stmt_name[0] != '\0') {
#ifdef ENABLE_MULTIPLE_NODES
psrc = IS_PGXC_DATANODE ? u_sess->pcache_cxt.cur_stmt_psrc : pstmt->plansource;
u_sess->pcache_cxt.gpc_in_batch = true;
#else
psrc = pstmt->plansource;
#endif
@ -10984,6 +11037,8 @@ static void exec_batch_bind_execute(StringInfo input_message)
pfree(completionTag);
}
/* end batch, reset gpc batch flag */
u_sess->pcache_cxt.gpc_in_batch = false;
/* Done with the snapshot used */
if (snapshot_set)

3
src/gausskernel/process/threadpool/knl_session.cpp
View File

@ -425,6 +425,8 @@ static void knl_u_plancache_init(knl_u_plancache_context* pcache_cxt)
pcache_cxt->query_has_params = false;
pcache_cxt->prepared_queries = NULL;
pcache_cxt->lightproxy_objs = NULL;
pcache_cxt->stmt_lightproxy_htab = NULL;
pcache_cxt->unnamed_gpc_lp = NULL;
pcache_cxt->datanode_queries = NULL;
pcache_cxt->unnamed_stmt_psrc = NULL;
pcache_cxt->cur_stmt_psrc = NULL;
@ -435,6 +437,7 @@ static void knl_u_plancache_init(knl_u_plancache_context* pcache_cxt)
pcache_cxt->gpc_remote_msg = false;
pcache_cxt->gpc_first_send = true;
pcache_cxt->gpc_in_try_store = false;
pcache_cxt->gpc_in_batch = false;
}
static void knl_u_typecache_init(knl_u_typecache_context* tycache_cxt)

7
src/gausskernel/process/threadpool/threadpool_scheduler.cpp
View File

@ -53,9 +53,9 @@ static void SchedulerSIGKILLHandler(SIGNAL_ARGS)
proc_exit(0);
}
static void SchedulerSIGHUPHandler(SIGNAL_ARGS)
void ThreadPoolScheduler::SigHupHandler()
{
t_thrd.threadpool_cxt.scheduler->m_getSIGHUP = true;
m_getSIGHUP = true;
}
static void reloadConfigFileIfNecessary()
@ -70,7 +70,6 @@ void TpoolSchedulerMain(ThreadPoolScheduler *scheduler)
{
int gpc_count = 0;
(void)gspqsignal(SIGHUP, SchedulerSIGHUPHandler);
(void)gspqsignal(SIGKILL, SchedulerSIGKILLHandler);
gs_signal_setmask(&t_thrd.libpq_cxt.UnBlockSig, NULL);
(void)gs_signal_unblock_sigusr2();
@ -130,7 +129,7 @@ void ThreadPoolScheduler::DynamicAdjustThreadPool()
void ThreadPoolScheduler::GPCScheduleCleaner(int* gpc_count)
{
if (ENABLE_GPC && *gpc_count == GPC_CLEAN_TIME) {
if (pmState == PM_RUN) {
if (pmState == PM_RUN || pmState == PM_HOT_STANDBY) {
MemoryContext oldCxt = MemoryContextSwitchTo(m_gpcContext);
pthread_mutex_lock(&g_instance.gpc_reset_lock);
g_instance.plan_cache->DropInvalid();

2
src/gausskernel/process/threadpool/threadpool_worker.cpp
View File

@ -616,6 +616,8 @@ void ThreadPoolWorker::CleanUpSession(bool threadexit)
if (!t_thrd.proc_cxt.proc_exit_inprogress) {
if (ENABLE_DN_GPC)
CleanSessGPCPtr(m_currentSession);
if (u_sess->pcache_cxt.unnamed_stmt_psrc && u_sess->pcache_cxt.unnamed_stmt_psrc->gpc.status.InShareTable())
u_sess->pcache_cxt.unnamed_stmt_psrc->gpc.status.SubRefCount();
CNGPCCleanUpSession();
}

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

@ -69,7 +69,7 @@
const int NUM_PERCENTILE_COUNT = 2;
const int INIT_NUMA_ALLOC_COUNT = 32;
const int HOTKEY_ABANDON_LENGTH = 100;
const int MAX_GLOBAL_CACHEMEM_NUM = 8;
const int MAX_GLOBAL_CACHEMEM_NUM = 128;
enum knl_virtual_role {
VUNKNOWN = 0,
@ -887,7 +887,8 @@ extern void add_numa_alloc_info(void* numaAddr, size_t length);
#define ATOMIC_TRUE 1
#define ATOMIC_FALSE 0
#define GLOBAL_PLANCACHE_MEMCONTEXT (g_instance.cache_cxt.global_plancache_mem[random() % MAX_GLOBAL_CACHEMEM_NUM])
#define GLOBAL_PLANCACHE_MEMCONTEXT \
(g_instance.cache_cxt.global_plancache_mem[u_sess->session_id % MAX_GLOBAL_CACHEMEM_NUM])
#endif /* SRC_INCLUDE_KNL_KNL_INSTANCE_H_ */

5
src/include/knl/knl_session.h
View File

@ -769,7 +769,9 @@ typedef struct knl_u_plancache_context {
*/
HTAB* prepared_queries;
HTAB* stmt_lightproxy_htab;
HTAB* stmt_lightproxy_htab; /* mapping statement name and lightproxy obj, only for gpc */
lightProxy* unnamed_gpc_lp; /* light proxy ptr for shard unnamed cachedplansource, only for gpc */
HTAB* lightproxy_objs;
@ -793,6 +795,7 @@ typedef struct knl_u_plancache_context {
bool gpc_remote_msg;
bool gpc_first_send;
bool gpc_in_try_store;
bool gpc_in_batch; /* true if is doing 2 ~ n batch execute, false if not in batch or doing first batch execute */
} knl_u_plancache_context;
typedef struct knl_u_typecache_context {

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

@ -31,6 +31,7 @@ public:
~ThreadPoolScheduler();
int StartUp();
void DynamicAdjustThreadPool();
void SigHupHandler();
void GPCScheduleCleaner(int* gpc_count);
void ShutDown() const;
inline ThreadId GetThreadId()

5
src/include/utils/globalplancache.h
View File

@ -52,6 +52,7 @@ public:
void RemoveEntry(uint32 htblIdx, GPCEntry *entry);
template<PlansourceInvalidAction action_type>
void RemovePlanSource(CachedPlanSource* plansource, const char* stmt_name);
void MoveIntoInvalidPlanList(CachedPlanSource* psrc);
bool TryStore(CachedPlanSource *plansource, PreparedStatement *ps);
Datum PlanClean();
void CleanUpByTime();
@ -63,9 +64,9 @@ public:
/* transaction */
void RecreateCachePlan(CachedPlanSource* oldsource, const char* stmt_name,
PreparedStatement *entry, SPIPlanPtr spiplan, ListCell* spiplanCell);
PreparedStatement *entry, SPIPlanPtr spiplan, ListCell* spiplanCell, bool hasGetLock);
void Commit();
bool CheckRecreateCachePlan(CachedPlanSource* psrc);
bool CheckRecreateCachePlan(CachedPlanSource* psrc, bool* hasGetLock);
void CNCommit();
void DNCommit();

2
src/include/utils/plancache.h
View File

@ -415,5 +415,7 @@ extern void PlanCacheRelCallback(Datum arg, Oid relid);
extern void PlanCacheFuncCallback(Datum arg, int cacheid, uint32 hashvalue);
extern void PlanCacheSysCallback(Datum arg, int cacheid, uint32 hashvalue);
extern bool IsStreamSupport();
extern void AcquirePlannerLocks(List* stmt_list, bool acquire);
extern void AcquireExecutorLocks(List* stmt_list, bool acquire);
#endif /* PLANCACHE_H */