diff --git a/src/common/backend/utils/cache/inval.cpp b/src/common/backend/utils/cache/inval.cpp index 6ec3d972..cce6496a 100644 --- a/src/common/backend/utils/cache/inval.cpp +++ b/src/common/backend/utils/cache/inval.cpp @@ -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; } } diff --git a/src/common/backend/utils/cache/plancache.cpp b/src/common/backend/utils/cache/plancache.cpp index 62b3e133..3cff094f 100644 --- a/src/common/backend/utils/cache/plancache.cpp +++ b/src/common/backend/utils/cache/plancache.cpp @@ -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; diff --git a/src/gausskernel/optimizer/commands/prepare.cpp b/src/gausskernel/optimizer/commands/prepare.cpp index c95b7d38..bc8a2d28 100644 --- a/src/gausskernel/optimizer/commands/prepare.cpp +++ b/src/gausskernel/optimizer/commands/prepare.cpp @@ -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); diff --git a/src/gausskernel/process/globalplancache/globalplancache.cpp b/src/gausskernel/process/globalplancache/globalplancache.cpp index af2ee2d0..2005ffec 100644 --- a/src/gausskernel/process/globalplancache/globalplancache.cpp +++ b/src/gausskernel/process/globalplancache/globalplancache.cpp @@ -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(oldsource, stmt_name); } diff --git a/src/gausskernel/process/globalplancache/globalplancache_view.cpp b/src/gausskernel/process/globalplancache/globalplancache_view.cpp index 1307d467..ae55088e 100644 --- a/src/gausskernel/process/globalplancache/globalplancache_view.cpp +++ b/src/gausskernel/process/globalplancache/globalplancache_view.cpp @@ -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 && diff --git a/src/gausskernel/process/postmaster/postmaster.cpp b/src/gausskernel/process/postmaster/postmaster.cpp index 234adb24..d58ac6da 100755 --- a/src/gausskernel/process/postmaster/postmaster.cpp +++ b/src/gausskernel/process/postmaster/postmaster.cpp @@ -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) diff --git a/src/gausskernel/process/tcop/postgres.cpp b/src/gausskernel/process/tcop/postgres.cpp index 409b0955..1151c3ec 100644 --- a/src/gausskernel/process/tcop/postgres.cpp +++ b/src/gausskernel/process/tcop/postgres.cpp @@ -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) diff --git a/src/gausskernel/process/threadpool/knl_session.cpp b/src/gausskernel/process/threadpool/knl_session.cpp index 721ecee6..fa6a298b 100644 --- a/src/gausskernel/process/threadpool/knl_session.cpp +++ b/src/gausskernel/process/threadpool/knl_session.cpp @@ -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) diff --git a/src/gausskernel/process/threadpool/threadpool_scheduler.cpp b/src/gausskernel/process/threadpool/threadpool_scheduler.cpp index 75f3cd02..30e020ab 100644 --- a/src/gausskernel/process/threadpool/threadpool_scheduler.cpp +++ b/src/gausskernel/process/threadpool/threadpool_scheduler.cpp @@ -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(); diff --git a/src/gausskernel/process/threadpool/threadpool_worker.cpp b/src/gausskernel/process/threadpool/threadpool_worker.cpp index 83e15316..001f052d 100644 --- a/src/gausskernel/process/threadpool/threadpool_worker.cpp +++ b/src/gausskernel/process/threadpool/threadpool_worker.cpp @@ -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(); } diff --git a/src/include/knl/knl_instance.h b/src/include/knl/knl_instance.h index 3b7d9a35..9cc51d04 100644 --- a/src/include/knl/knl_instance.h +++ b/src/include/knl/knl_instance.h @@ -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_ */ diff --git a/src/include/knl/knl_session.h b/src/include/knl/knl_session.h index cd73003d..dba1fa6c 100644 --- a/src/include/knl/knl_session.h +++ b/src/include/knl/knl_session.h @@ -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 { diff --git a/src/include/threadpool/threadpool_scheduler.h b/src/include/threadpool/threadpool_scheduler.h index 7dd032bb..be2e63e5 100644 --- a/src/include/threadpool/threadpool_scheduler.h +++ b/src/include/threadpool/threadpool_scheduler.h @@ -31,6 +31,7 @@ public: ~ThreadPoolScheduler(); int StartUp(); void DynamicAdjustThreadPool(); + void SigHupHandler(); void GPCScheduleCleaner(int* gpc_count); void ShutDown() const; inline ThreadId GetThreadId() diff --git a/src/include/utils/globalplancache.h b/src/include/utils/globalplancache.h index 2749fbf4..887873df 100644 --- a/src/include/utils/globalplancache.h +++ b/src/include/utils/globalplancache.h @@ -52,6 +52,7 @@ public: void RemoveEntry(uint32 htblIdx, GPCEntry *entry); template 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(); diff --git a/src/include/utils/plancache.h b/src/include/utils/plancache.h index aa725a3b..7fcc5f47 100644 --- a/src/include/utils/plancache.h +++ b/src/include/utils/plancache.h @@ -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 */