Google Git
Sign in
apache / hawq / 1313a2e67e33a9817a5226ef190a6b4c143fb278 / . / src / backend / resourcemanager / resourcemanager.c
blob: 7303382f53acdb7faf8851915bfc1d4f9541b424 [file] [log] [blame]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
#include "envswitch.h"
#include "dynrm.h"
#include "utils/hashtable.h"
#include "resourcemanager/resourcemanager.h"
#include "resourceenforcer/resourceenforcer.h"
#include "storage/ipc.h"
#include "postmaster/postmaster.h"
#include "postmaster/fork_process.h"
#include "postmaster/syslogger.h"
#include "utils/kvproperties.h"
#include "utils/network_utils.h"
#include "miscadmin.h"
#include "communication/rmcomm_QD2RM.h"
#include "communication/rmcomm_AsyncComm.h"
#include "communication/rmcomm_MessageHandler.h"
#include "communication/rmcomm_MessageServer.h"
#include "communication/rmcomm_RM2GRM_libyarn.h"
#include "communication/rmcomm_RM2GRM_none.h"
#include "communication/rmcomm_RMSEG2RM.h"
#include "communication/rmcomm_RM2RMSEG.h"
#include "storage/proc.h"
#include "catalog/pg_database.h"
#include "catalog/pg_tablespace.h"
#include "catalog/catalog.h"
#include "utils/syscache.h"
#include "storage/backendid.h"
#include "storage/sinvaladt.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_resqueue.h"
#include "access/htup.h"
#include "utils/tqual.h"
#include "access/xact.h"
#include "resourcebroker/resourcebroker_API.h"
#include <executor/spi.h>
#include "gp-libpq-fe.h"
#include "gp-libpq-int.h"
extern bool FindMyDatabase(const char *name, Oid *db_id, Oid *db_tablespace);
static char *probeDatabase = "template1";
int loadAllQueueAndUser(void);
int loadHostInformationIntoResourcePool(void);
/** Functions for signal handling. **/
void quitResManager(SIGNAL_ARGS);
void handleChildSignal(SIGNAL_ARGS);
void notifyPostmasterResManagerStarted(SIGNAL_ARGS);
int generateAllocRequestToBroker(void);
void completeAllocRequestToBroker(int32_t *reqmem,
int32_t *reqcore,
List **preferred);
void processResourceBrokerTasks(void);
void generateResourceRequestToResourceBroker(void);
void cleanupAllGRMContainers(void);
bool cleanedAllGRMContainers(void);
/** Functions for loading resource queue and user definition. **/
#define DEC_VAR_PG_CATCOL(name) Datum name##Datum; bool name##IsNull = false;
#define GETATTR_PG_CATCOL(tbname,colname,relvar,tupvar) \
colname##Datum = heap_getattr(tupvar, \
Anum_##tbname##_##colname, \
relvar->rd_att, \
&colname##IsNull);
#define BLDPROP_PG_CATCOL(colname,list,tag1,attrtag,attrid,attrtyp,index) \
insertDQueueTailNode(list, \
createProperty##attrtyp(list->Context, \
tag1, \
get##attrtag##AttributeName(attrid), \
&index, \
DatumGet##attrtyp(colname##Datum)));
#define BLDPROP_PG_CATCOL_OID(colname,list,tag1,attrtag,attrid,index) \
insertDQueueTailNode(list, \
createPropertyOID(list->Context, \
tag1, \
get##attrtag##AttributeName(attrid), \
&index, \
colname##IsNull ? InvalidOid : \
DatumGetObjectId(colname##Datum)));
#define BLDPROP_PG_CATCOL_SOID(list,tag1,attrtag,attrid,index,oid) \
insertDQueueTailNode(list, \
createPropertyOID(list->Context, \
tag1, \
get##attrtag##AttributeName(attrid), \
&index, \
oid));
#define BLDPROP_PG_CATCOL_SET(list,tag1,attrtag,attrid,attrtyp,index,val) \
insertDQueueTailNode(list, \
createProperty##attrtyp(list->Context, \
tag1, \
get##attrtag##AttributeName(attrid), \
&index, \
val));
#define BLDPROP_PG_CATCOL_TXT(colname,list,tag1,attrtag,attrid,index) \
insertDQueueTailNode(list, \
createPropertyText(list->Context, \
tag1, \
get##attrtag##AttributeName(attrid), \
&index, \
colname##IsNull ? NULL : \
DatumGetTextP(colname##Datum)));
#define BLDPROP_PG_CATCOL_FLOAT(colname,list,tag1,attrtag,attrid,index) \
insertDQueueTailNode(list, \
createPropertyFloat(list->Context, \
tag1, \
get##attrtag##AttributeName(attrid), \
&index, \
colname##IsNull ? -1.0 : \
DatumGetFloat4(colname##Datum)));
#define BLDPROP_PG_CATCOL_STR(list,tag1,attrtag,attrid,index,val) \
insertDQueueTailNode(list, \
createPropertyString(list->Context, \
tag1, \
get##attrtag##AttributeName(attrid), \
&index, \
val));
#define BLDPROP_PG_CATCOL_INT32(colname,list,tag1,attrtag,attrid,index) \
insertDQueueTailNode(list, \
createPropertyInt32(list->Context, \
tag1, \
get##attrtag##AttributeName(attrid), \
&index, \
colname##IsNull ? -1 : \
DatumGetInt32(colname##Datum)));
#define BLDPROP_PG_CATCOL_INT64(colname,list,tag1,attrtag,attrid,index) \
insertDQueueTailNode(list, \
createPropertyInt32(list->Context, \
tag1, \
get##attrtag##AttributeName(attrid), \
&index, \
colname##IsNull ? -1 : \
DatumGetInt64(colname##Datum)));
int loadUserPropertiesFromCatalog(List **users);
int loadQueuePropertiesFromCatalog(List **queues);
/******************************************************************************
* The global instance of dynamic resource manager.
******************************************************************************/
DynRMGlobal DRMGlobalInstance;/* The global instance for HAWQ RM logic. */
volatile bool DRMStartup; /* This variable is set only in postmaster
process by signal handler SIG_USR2. This is
used to check if HAWQ RM is ready to accept
request. */
pthread_t t_move_cgroup;
queue *g_queue_cgroup;
GHash g_ghash_cgroup;
uint64 rm_enforce_last_cleanup_time;
volatile bool g_enforcement_thread_quited = false;
#define SEGMENT_HEARTBEAT_INTERVAL (3LL * 1000000LL)
/***************************
* The main entry of HAWQ RM
***************************/
int ResManagerMain(int argc, char *argv[])
{
int res = FUNC_RETURN_OK;
/*******************************************************************/
/* Step 1. Prepare memory context, log and create global instance. */
/*******************************************************************/
IsUnderPostmaster = true;
MyProcPid = getpid();
SetProcessingMode(InitProcessing);
res = createDRMInstance();
if ( res != FUNC_RETURN_OK ) {
elog(FATAL, "HAWQ RM Can not create resource manager global instance.");
}
elog(DEBUG5, "HAWQ RM :: created dynamic resource manager instance.");
res = createDRMMemoryContext();
if ( res != FUNC_RETURN_OK ) {
elog(FATAL, "HAWQ RM Can not create resource manager global instance.");
}
elog(DEBUG5, "HAWQ RM :: created resource manager memory context.");
res = initializeDRMInstance(PCONTEXT);
if ( res != FUNC_RETURN_OK ) {
elog(FATAL, "HAWQ RM Can not initialize global instance.");
}
elog(DEBUG5, "HAWQ RM :: initialized resource manager instance.");
/**************************************************************************
* STEP 2. Load configuration.
**************************************************************************/
/* Command line parser. */
res = parseCommandLine(argc, argv);
if ( res != FUNC_RETURN_OK ) {
elog(FATAL, "Wrong resource manager command line arguments.");
}
elog(DEBUG5, "HAWQ RM ::passed command line arguments.");
/* Recognize all loaded configure properties. */
res = loadDynamicResourceManagerConfigure();
if ( res != FUNC_RETURN_OK ) {
elog(FATAL, "Fail to load valid properties from configure files.");
}
elog(DEBUG5, "HAWQ RM :: passed loading configuration.");
/*******************************************************/
/* Step 3. Post-initialization of core data structure. */
/*******************************************************/
initializeDRMCore();
elog(DEBUG5, "HAWQ RM :: Passed initializing core data structure.");
/**************************************************************************/
/* STEP 4. INIT for making RM process access catalog by CAQL etc. */
/**************************************************************************/
/* BLOCK signal behavior. Only another specific thread has the capability to
* process the signal. */
PG_SETMASK(&BlockSig);
pqsignal(SIGHUP , SIG_IGN);
pqsignal(SIGINT , quitResManager);
pqsignal(SIGTERM, quitResManager);
pqsignal(SIGQUIT, quitResManager);
pqsignal(SIGPIPE, SIG_IGN);
pqsignal(SIGUSR1, SIG_IGN);
pqsignal(SIGUSR2, quitResManager);
pqsignal(SIGCHLD, handleChildSignal);
pqsignal(SIGTTIN, SIG_IGN);
pqsignal(SIGTTOU, SIG_IGN);
if ( DRMGlobalInstance->Role == START_RM_ROLE_MASTER ) {
CurrentResourceOwner = ResourceOwnerCreate(NULL, "Resource Manager");
BaseInit();
InitProcess();
InitBufferPoolBackend();
InitXLOGAccess();
SetProcessingMode(NormalProcessing);
MyDatabaseId = TemplateDbOid;
MyDatabaseTableSpace = DEFAULTTABLESPACE_OID;
if (!FindMyDatabase(probeDatabase, &MyDatabaseId, &MyDatabaseTableSpace))
ereport(FATAL, (errcode(ERRCODE_UNDEFINED_DATABASE),
errmsg("database 'postgres' does not exist")));
char *fullpath = GetDatabasePath(MyDatabaseId, MyDatabaseTableSpace);
SetDatabasePath(fullpath);
InitProcessPhase2();
MyBackendId = InvalidBackendId;
SharedInvalBackendInit(false);
if (MyBackendId > MaxBackends || MyBackendId <= 0)
elog(FATAL, "bad backend id: %d", MyBackendId);
InitBufferPoolBackend();
RelationCacheInitialize();
InitCatalogCache();
RelationCacheInitializePhase2();
}
/* END: INIT for making RM process access catalog by caql etc. */
/**************************************************************************/
PG_SETMASK(&UnBlockSig);
/* Save process fork mode. */
DRMGlobalInstance->ThisPID = getpid();
DRMGlobalInstance->ParentPID = getppid();
/* Initialize socket connection pool. */
initializeSocketConnectionPool();
elog(DEBUG5, "HAWQ RM :: starts as role %d.", DRMGlobalInstance->Role);
/*******************************************/
/* STEP 5. Start concrete processing loop. */
/*******************************************/
switch(DRMGlobalInstance->Role) {
case START_RM_ROLE_MASTER:
{
/* Mark as master side resource manager. */
int oldid = gp_session_id;
gp_session_id = -1;
init_ps_display("master resource manager", " ", "", "");
gp_session_id = oldid;
/* Initialize resource broker service. */
RB_prepareImplementation(DRMGlobalInstance->ImpType);
res = RB_start(true);
if ( res != FUNC_RETURN_OK ) {
elog(FATAL, "Fail to create resource broker service.");
}
res = ResManagerMainServer2ndPhase();
elog(LOG, "Master RM exits.\n");
break;
}
case START_RM_ROLE_SEGMENT:
{
init_ps_display("segment resource manager", "", "","");
res = ResManagerMainSegment2ndPhase();
elog(LOG, "Segment RM exits.\n");
break;
}
default:
Assert(false); /* Should never come here. */
}
return res;
}
/*
* Start sub-process as resource manager from postmaster.
*
* rmrole[in] Specify the role of RM process, possible values are :
* START_RM_ROLE_SERVER : RM server
* START_RM_ROLE_SEGMENT: RM agent.
*/
int ResManagerProcessStartup(void)
{
pid_t ResMgrPID = 0;
char *arguments[3] = {NULL, NULL, NULL};
arguments[0] = "postgres";
arguments[1] = "-role";
int rmrole = AmIMaster() ? START_RM_ROLE_MASTER : START_RM_ROLE_SEGMENT;
// if in upgrade mode, at the first calling, the old pg_resqueue data can't let resource manager works,
// And also we only need to upgrade master node, no segment is enabled at this time, so no need for resource manager.
if(gp_upgrade_mode)
return 0;
/*
* Prepare temporary signal handler for SIGUSR2 to let postmaster process
* able to know when HAWQ RM is ready to accept requests.
*/
DRMStartup = false;
pqsignal(SIGUSR2, notifyPostmasterResManagerStarted);
PG_SETMASK(&UnBlockSig);
/* Fork resource manager process. */
switch ((ResMgrPID = fork_process()))
{
case -1:
pqsignal(SIGUSR2, SIG_IGN);
PG_SETMASK(&BlockSig);
ereport(FATAL,
(errmsg("could not fork resource manager collector: %m")));
return 0;
case 0:
pqsignal(SIGUSR2, SIG_IGN);
/* in postmaster child ... */
/* Close the postmaster's sockets */
ClosePostmasterPorts(false);
/* Lose the postmaster's on-exit routines */
on_exit_reset();
ResourceManagerIsForked = true;
switch(rmrole) {
case START_RM_ROLE_MASTER:
arguments[2] = "server";
proc_exit(ResManagerMain(3, arguments));
break;
case START_RM_ROLE_SEGMENT:
arguments[2] = "segment";
proc_exit(ResManagerMain(3, arguments));
break;
}
break;
default:
{
/*******************************************************************
* IN POSTMASTER. Postmaster here waits for the signal SIGUSR2 sent
* from HAWQ RM to gurantee that HAWQ RM can accept the request after
* the return of this function.
******************************************************************/
/* wait to ensure that resource manager works basically. */
for ( int i = 0 ; i < 180 ; ++i ) {
if ( usleep(1000000) != 0 )
i--;
elog(LOG, "Wait for HAWQ RM %d", i);
if ( DRMStartup )
break;
}
pqsignal(SIGUSR2, SIG_IGN);
PG_SETMASK(&BlockSig);
if ( DRMStartup ) {
elog(LOG, "HAWQ :: Received signal notification that HAWQ RM "
"works now.");
}
else {
elog(FATAL, "HAWQ RM can not work. Please check HAWQ RM log.");
}
return ResMgrPID;
}
}
/* shouldn't get here */
return 0;
}
int ResManagerMainServer2ndPhase(void)
{
int res = FUNC_RETURN_OK;
/* Register message handlers */
registerMessageHandler(REQUEST_QD_CONNECTION_REG , handleRMRequestConnectionReg);
registerMessageHandler(REQUEST_QD_CONNECTION_REG_OID , handleRMRequestConnectionRegByOID);
registerMessageHandler(REQUEST_QD_CONNECTION_UNREG , handleRMRequestConnectionUnReg);
registerMessageHandler(REQUEST_QD_ACQUIRE_RESOURCE , handleRMRequestAcquireResource);
registerMessageHandler(REQUEST_QD_RETURN_RESOURCE , handleRMRequestReturnResource);
registerMessageHandler(REQUEST_RM_IMALIVE , handleRMSEGRequestIMAlive);
registerMessageHandler(REQUEST_QD_DDL_MANIPULATERESQUEUE, handleRMDDLRequestManipulateResourceQueue);
registerMessageHandler(REQUEST_QD_DDL_MANIPULATEROLE , handleRMDDLRequestManipulateRole);
registerMessageHandler(REQUEST_QD_ACQUIRE_RESOURCE_QUOTA, handleRMRequestAcquireResourceQuota);
registerMessageHandler(REQUEST_QD_REFRESH_RESOURCE , handleRMRequestRefreshResource);
registerMessageHandler(REQUEST_QD_SEGMENT_ISDOWN , handleRMRequestSegmentIsDown);
registerMessageHandler(REQUEST_QD_DUMP_STATUS , handleRMRequestDumpStatus);
registerMessageHandler(REQUEST_QD_DUMP_RESQUEUE_STATUS , handleRMRequestDumpResQueueStatus);
registerMessageHandler(REQUEST_DUMMY , handleRMRequestDummy);
registerMessageHandler(REQUEST_QD_QUOTA_CONTROL , handleRMRequestQuotaControl);
/* New socket facility poll based server.*/
res = initializeSocketServer();
if ( res != FUNC_RETURN_OK ) {
elog(FATAL, "Fail to initialize socket server.");
}
/*
* Notify postmaster that HAWQ RM is ready. Ignore the possible problem that
* the parent process quits. HAWQ RM will automatically detect if its parent
* dies, then HAWQ RM should exit normally.
*/
kill(DRMGlobalInstance->ParentPID, SIGUSR2);
/* Clean up gp_segment_configuration table. */
cleanup_segment_config();
cleanup_segment_config_history();
/*
* register itself into gp_segment_configuration table
* master internal id is 0, segment id starts from 1
*/
add_segment_config_row(MASTER_ORDER_ID,
DRMGlobalInstance->SocketLocalHostName.Str,
DRMGlobalInstance->SocketLocalHostName.Str,
PostPortNumber,
SEGMENT_ROLE_MASTER_CONFIG,
SEGMENT_STATUS_UP,
"");
/* Load queue and user definition as no DDL now. */
res = loadAllQueueAndUser();
if ( res != FUNC_RETURN_OK ) {
elog(FATAL, "Fail to load queue and user definition.");
}
elog(DEBUG5, "HAWQ RM :: passed loading queue and user definition.");
/* Check slaves file firstly to ensure we have expected cluster size. */
checkSlavesFile();
if ( rm_resourcepool_test_filename != NULL &&
rm_resourcepool_test_filename[0] != '\0' ) {
loadHostInformationIntoResourcePool();
}
/******* TILL NOW, resource manager starts providing services *******/
elog(LOG, "HAWQ RM process works now.");
/* Start request handler to provide services. */
res = MainHandlerLoop();
/* res is returned to the caller. */
/******* TILL NOW, resource manager goes into EXIT PHASE *******/
elog(LOG, "HAWQ RM server goes into exit phase.");
/* Shutdown resource broker. */
RB_stop();
return res;
}
/**
* The main loop of HAWQ RM process.
*/
int MainHandlerLoop(void)
{
int res = FUNC_RETURN_OK;
DRMGlobalInstance->ResourceManagerStartTime = gettime_microsec();
while( DRMGlobalInstance->ResManagerMainKeepRun )
{
/* STEP 0. Check if postmaster process exists. */
if (!PostmasterIsAlive(true)) {
DRMGlobalInstance->ResManagerMainKeepRun = false;
elog(LOG, "Postmaster is not alive, resource manager exits");
break;
}
/* STEP 1. Check resource broker status. */
RB_start(true);
/* STEP 2. Check if resource manager is in clean up status. */
if ( isCleanGRMResourceStatus() )
{
/*
* Fisrtly we move all currently accepted GRM containers into resource
* pool. We will drop them at once.
*/
moveAllAcceptedGRMContainersToResPool();
/*
* Mark all current active connection tracks that they are in fact
* using old application's resource. When resource is returned, the
* resource is calculated in old resource level.
*/
setAllAllocatedResourceInConnectionTracksOld();
/* Validate that all in-use resource are marked to old resource. */
Assert(isAllResourceQueueIdle());
/*
* Reset deadlock detector. The deadlock detector works for new
* resource only.
*/
resetAllDeadLockDetector();
/*
* Mark all segments DOWN due to no GRM cluster report
*/
setAllNodesGRMDown();
/* Move all resource broker allocated GRM containers to returned. */
cleanupAllGRMContainers();
/* Refresh resource queue resource usage and request quota. */
refreshMemoryCoreRatioLevelUsage(gettime_microsec());
resetAllSegmentsGRMContainerFailAllocCount();
resetAllSegmentsNVSeg();
/* Check if can resume using new available global resource manager.*/
if ( cleanedAllGRMContainers() )
{
unsetCleanGRMResourceStatus();
}
}
/*
* STEP 3. Process all communications between resource manager and
* resource broker.
*/
processResourceBrokerTasks();
/* STEP 4. Handle socket server inputs. */
res = processAllCommFileDescs();
if ( res != FUNC_RETURN_OK )
{
/*
* The possible error here is the failure of poll(), we won't keep
* running HAWQ RM any longer, graceful quit is requested.
*/
elog(WARNING, "System error cause resource manager not possible to "
"track network communications. Exit resource manager "
"process normally.");
DRMGlobalInstance->ResManagerMainKeepRun = false;
}
/* STEP 5. Handle all submitted requests through socket clients. */
processSubmittedRequests();
/* STEP 6. Generate possible resource request to resource broker. */
generateResourceRequestToResourceBroker();
/* STEP 7. Check timeout resource allocation and timeout queuing requests. */
timeoutDeadResourceAllocation();
timeoutQueuedRequest();
/*
* STEP 8. Check the status of all segment nodes, mark down if hasn't got
* IMAlive message for a pre-defined period.
*/
uint64_t curtime = gettime_microsec();
if ((rm_resourcepool_test_filename == NULL ||
rm_resourcepool_test_filename[0] == '\0') &&
(curtime - PRESPOOL->LastCheckTime >
1000000LL * rm_segment_heartbeat_timeout))
{
updateStatusOfAllNodes();
PRESPOOL->LastCheckTime = curtime;
}
/* STEP 9. Move all accepted GRM containers into resource pool. */
moveAllAcceptedGRMContainersToResPool();
/*
* STEP 10. Check if should pause dispatching resource to queries to
* collect resource back in order to return GRM containers.
*/
if ( DRMGlobalInstance->ImpType != NONE_HAWQ2 &&
PRESPOOL->AddPendingContainerCount == 0 &&
PRESPOOL->RetPendingContainerCount == 0 &&
PQUEMGR->ForcedReturnGRMContainerCount == 0 &&
PQUEMGR->GRMQueueCurCapacity > 1 &&
PQUEMGR->GRMQueueResourceTight )
{
elog(LOG, "Resource manager decides to breathe out resource. "
"Current relative GRM queue capacity %lf, "
"Expect GRM queue capacity %lf, "
"Estimae GRM queue %s",
PQUEMGR->GRMQueueCurCapacity,
PQUEMGR->GRMQueueCapacity,
PQUEMGR->GRMQueueResourceTight ? "busy" : "not busy");
/* Calculate how many GRM containers should be returned. */
setForcedReturnGRMContainerCount();
}
/* STEP 11. Dispatch resource to queries and send the messages out.*/
if ( PRESPOOL->Segments.NodeCount > 0 && PQUEMGR->RatioCount > 0 &&
PQUEMGR->toRunQueryDispatch &&
PQUEMGR->ForcedReturnGRMContainerCount == 0 &&
PRESPOOL->SlavesHostCount > 0 )
{
dispatchResourceToQueries();
}
else if ( PQUEMGR->ForcedReturnGRMContainerCount > 0 )
{
forceReturnGRMResourceToRB();
}
else
{
elog(DEBUG3, "Loop dummy. %d, %d, %d, %d",
PRESPOOL->Segments.NodeCount,
PQUEMGR->RatioCount,
PQUEMGR->toRunQueryDispatch ? 1 : 0,
PQUEMGR->ForcedReturnGRMContainerCount);
}
/* STEP 12. Generate output content to client connections. */
sendResponseToClients();
/*
* STEP 13. Return containers to global resource manager if there some
* some idle resource.
*/
timeoutIdleGRMResourceToRB();
/* STEP 14. Notify segments to increase resource quota. */
notifyToBeAcceptedGRMContainersToRMSEG();
/* STEP 15. Notify segments to decrease resource. */
notifyToBeKickedGRMContainersToRMSEG();
/*
* STEP 16. Check slaves file if the content is not checked or is
* updated.
*/
checkSlavesFile();
}
elog(LOG, "Resource manager main event handler exits.");
return res;
}
/**
* Parse passed in command line arguments. The acceptable argument keys and
* values are all listed below which are locally used in parseCommandLine().
*/
/* -role + rolename */
#define HAWQDRM_COMMANDLINE_ROLE "-role"
#define HAWQDRM_COMMANDLINE_ROLE_VALUE_SERVER "server"
#define HAWQDRM_COMMANDLINE_ROLE_VALUE_SEGMENT "segment"
int parseCommandLine(int argc, char **argv)
{
int res = FUNC_RETURN_OK;
SimpString value;
initSimpleString(&value, PCONTEXT);
for ( int i = 0 ; i < argc ; ++i ) {
if ( strcmp(argv[i], HAWQDRM_COMMANDLINE_ROLE) == 0 ) {
res = getStringValue(argc, argv, i+1, &value);
if ( res != FUNC_RETURN_OK ) {
elog( WARNING,
"Wrong command line argument %s behind %s.",
value.Str,
HAWQDRM_COMMANDLINE_ROLE);
goto exit;
}
if ( SimpleStringComp(&value,
HAWQDRM_COMMANDLINE_ROLE_VALUE_SERVER) == 0 ) {
DRMGlobalInstance->Role = START_RM_ROLE_MASTER;
}
else if ( SimpleStringComp(&value,
HAWQDRM_COMMANDLINE_ROLE_VALUE_SEGMENT) == 0 ) {
DRMGlobalInstance->Role = START_RM_ROLE_SEGMENT;
}
else {
elog(WARNING, "Wrong argument value '%s' for %s",
value.Str, HAWQDRM_COMMANDLINE_ROLE);
res = MAIN_WRONG_COMMANDLINE;
}
elog(DEBUG5, "HAWQ RM :: Command line sets RM role: %d\n",
DRMGlobalInstance->Role);
}
}
exit:
freeSimpleStringContent(&value);
return res;
}
int getStringValue(int argc, char **argv, int pos, SimpString *val)
{
if ( pos >= argc )
return MAIN_WRONG_COMMANDLINE;
setSimpleStringNoLen(val, argv[pos]);
return FUNC_RETURN_OK;
}
int createDRMInstance(void)
{
int res = FUNC_RETURN_OK;
/* Create dynamic resource manager global instance holding all data. */
DRMGlobalInstance = NULL;
DRMGlobalInstance = (DynRMGlobal)
rm_palloc0(TopMemoryContext,
sizeof(struct DynRMGlobalData));
Assert( DRMGlobalInstance != NULL );
return res;
}
/*
* Initialize the dynamic resource manager instance. Return 0 if succeed, other-
* wise, -1 is returned. Resource manager's memory context is created as one
* child context which is referenced by DRMGlobalInstance->Context.
*
* TopMemoryContext ( global instance is allocated here )
* |
* V
* Resource manager context ( all consequent objects are allocated here )
*
*/
int createDRMMemoryContext(void)
{
int res = FUNC_RETURN_OK;
Assert( DRMGlobalInstance != NULL );
/* Create dynamic resource manager memory context,the new memory context is
created as the child of top memory context. */
DRMGlobalInstance->Context = NULL;
DRMGlobalInstance->Context = AllocSetContextCreate(
TopMemoryContext,
DRMGLOBAL_MEMORY_CONTEXT_NAME,
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE );
Assert( PCONTEXT != NULL );
return res;
}
int initializeDRMInstance(MCTYPE context)
{
int res = FUNC_RETURN_OK;
DRMGlobalInstance->ResourceQueueManager = NULL;
DRMGlobalInstance->ConnTrackManager = NULL;
DRMGlobalInstance->ResourcePoolInstance = NULL;
DRMGlobalInstance->ImpType = NONE_HAWQ2;
DRMGlobalInstance->LocalHostLastUpdateTime = 0;
DRMGlobalInstance->HeartBeatLastSentTime = 0;
DRMGlobalInstance->TmpDirLastCheckTime = 0;
DRMGlobalInstance->LocalHostStat = NULL;
initializeDQueue(&(DRMGlobalInstance->LocalHostTempDirectories), context);
DRMGlobalInstance->LocalHostFailedTmpDirList = NULL;
HASHCTL ctl;
ctl.keysize = sizeof(TmpDirKey);
ctl.entrysize = sizeof(TmpDirEntry);
ctl.hcxt = context;
/* Tell the working threads keep running. */
DRMGlobalInstance->ResManagerMainKeepRun = true;
/* Whether send IMAlive message */
DRMGlobalInstance->SendIMAlive = true;
/* Where to send the heart-beat. */
DRMGlobalInstance->SendToStandby = false;
DRMGlobalInstance->ResBrokerAppTimeStamp = 0;
DRMGlobalInstance->ResBrokerTriggerCleanup = false;
/* Get local host name here to make all components able to use this info. */
initSimpleString(&(DRMGlobalInstance->SocketLocalHostName), context);
res = getLocalHostName(&(DRMGlobalInstance->SocketLocalHostName));
if ( res != FUNC_RETURN_OK ) {
elog(WARNING, "Fail to get local host name.");
}
/* Set resource manager server startup time to 0, i.e. not started yet. */
DRMGlobalInstance->ResourceManagerStartTime = 0;
return res;
}
int initializeDRMInstanceForQD(void)
{
int res = FUNC_RETURN_OK;
/* Get local host name here to make all components able to use this info. */
initSimpleString(&(DRMGlobalInstance->SocketLocalHostName),
DRMGlobalInstance->Context);
res = getLocalHostName(&(DRMGlobalInstance->SocketLocalHostName));
if ( res != FUNC_RETURN_OK ) {
elog(WARNING, "Fail to get local host name.");
}
return res;
}
#define DRMQE2RM_MEMORY_CONTEXT_NAME "QE to RM communication"
bool QE2RM_Initialized = false;
MemoryContext QE2RM_CommContext = NULL;
void initializeDRMInstanceForQE(void)
{
if ( QE2RM_Initialized )
return;
int res = FUNC_RETURN_OK;
/* create dynamic resource manager instance to contain config data. */
res = createDRMInstance();
if ( res != FUNC_RETURN_OK ) {
elog(ERROR, "Fail to initialize data structure for communicating with "
"resource manager.");
}
MEMORY_CONTEXT_SWITCH_TO(TopMemoryContext)
QE2RM_CommContext = AllocSetContextCreate( CurrentMemoryContext,
DRMQE2RM_MEMORY_CONTEXT_NAME,
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE );
Assert( QE2RM_CommContext != NULL );
MEMORY_CONTEXT_SWITCH_BACK
DRMGlobalInstance->Context = QE2RM_CommContext;
/* Get local host name here to make all components able to use this info. */
initSimpleString(&(DRMGlobalInstance->SocketLocalHostName),
DRMGlobalInstance->Context);
res = getLocalHostName(&(DRMGlobalInstance->SocketLocalHostName));
if ( res != FUNC_RETURN_OK ) {
elog(WARNING, "Fail to get local host name.");
}
/****** Resource enforcement GUCs begins ******/
/* Get resource enforcement enablement flag */
DRMGlobalInstance->ResourceEnforcerCpuEnable = rm_enforce_cpu_enable;
/* Get resource enforcement CGroup mount point */
initSimpleString(&(DRMGlobalInstance->ResourceEnforcerCgroupMountPoint),
PCONTEXT);
setSimpleStringNoLen(&(DRMGlobalInstance->ResourceEnforcerCgroupMountPoint),
rm_enforce_cgrp_mnt_pnt);
/* Get resource enforcement CGroup hierarchy name */
initSimpleString(&(DRMGlobalInstance->ResourceEnforcerCgroupHierarchyName),
PCONTEXT);
setSimpleStringNoLen(&(DRMGlobalInstance->ResourceEnforcerCgroupHierarchyName),
rm_enforce_cgrp_hier_name);
/* Get resource enforcement mapping between hawq weight and yarn weight */
DRMGlobalInstance->ResourceEnforcerCpuWeight = rm_enforce_cpu_weight;
/* Get resource enforcement vcore-pcore ratio */
DRMGlobalInstance->ResourceEnforcerVcorePcoreRatio = rm_enforce_core_vpratio;
/* Get resource enforcement CGroup cleanup period */
DRMGlobalInstance->ResourceEnforcerCleanupPeriod = rm_enforce_cleanup_period;
/****** Resource enforcement GUCs ends ******/
QE2RM_Initialized = true;
}
/**
* Initialize resource manager core structure to get ready for services.
*/
void initializeDRMCore(void)
{
/* Initialize resource queue management part. */
DRMGlobalInstance->ResourceQueueManager =
(DynResourceQueueManager)
rm_palloc0(PCONTEXT, sizeof(DynResourceQueueManagerData));
initializeResourceQueueManager();
/* Initialize connection track manager part. */
DRMGlobalInstance->ConnTrackManager =
(ConnectionTrackManager)
rm_palloc0(PCONTEXT, sizeof(ConnectionTrackManagerData));
initializeConnectionTrackManager();
/* Initialize node resource management part. */
PRESPOOL = (ResourcePool) rm_palloc0(PCONTEXT, sizeof(ResourcePoolData));
initializeResourcePoolManager();
}
static void InitTemporaryDirs(DQueue tmpdirs_list, char *tmpdirs_string)
{
int templocation = -1;
SimpString tmpdirs;
initSimpleString(&tmpdirs, PCONTEXT);
setSimpleStringNoLen(&tmpdirs, tmpdirs_string);
while (FUNC_RETURN_OK == SimpleStringLocateChar(&tmpdirs, ',', &templocation))
{
SimpStringPtr tempdir = createSimpleString(PCONTEXT);
SimpleStringSubstring(&tmpdirs, 0, templocation, tempdir);
insertDQueueTailNode(tmpdirs_list, tempdir);
SimpString tmpstr;
initSimpleString(&tmpstr, PCONTEXT);
SimpleStringSubstring(&tmpdirs, templocation+1, -1, &tmpstr);
freeSimpleStringContent(&tmpdirs);
SimpleStringCopy(&tmpdirs, &tmpstr);
freeSimpleStringContent(&tmpstr);
}
SimpStringPtr tempdirlast = createSimpleString(PCONTEXT);
setSimpleStringWithContent(tempdirlast, tmpdirs.Str, tmpdirs.Len);
insertDQueueTailNode(tmpdirs_list, tempdirlast);
freeSimpleStringContent(&tmpdirs);
}
/**
* Read loaded config properties and fill the verified values into global
* instance.
*/
int loadDynamicResourceManagerConfigure(void)
{
elog(DEBUG3, "Resource manager loads Socket Listening Port %d",
rm_master_port);
elog(DEBUG3, "Resource manager loads Segment Socket Listening Port %d",
rm_segment_port);
/* Decide global resource manager mode. */
if ( strcasecmp(rm_global_rm_type, HAWQDRM_CONFFILE_SVRTYPE_VAL_YARN) == 0 )
{
DRMGlobalInstance->ImpType = YARN_LIBYARN;
}
else if ( strcasecmp(rm_global_rm_type, HAWQDRM_CONFFILE_SVRTYPE_VAL_NONE) == 0 )
{
DRMGlobalInstance->ImpType = NONE_HAWQ2;
}
else
{
elog(WARNING, "Wrong global resource manager type set in %s.",
HAWQDRM_CONFFILE_SERVER_TYPE);
return MAIN_CONF_UNSET_ROLE;
}
elog(DEBUG3, "Resource manager loads resource broker implement mode : %d",
DRMGlobalInstance->ImpType);
SimpString segmem;
if ( rm_seg_memory_use[0] == '\0' )
{
elog(WARNING, "%s is not set", HAWQDRM_CONFFILE_LIMIT_MEMORY_USE);
return MAIN_CONF_UNSET_SEGMENT_MEMORY_USE;
}
setSimpleStringRefNoLen(&segmem, rm_seg_memory_use);
int res = SimpleStringToStorageSizeMB(&segmem,
(uint32_t *) &(DRMGlobalInstance->SegmentMemoryMB));
if ( res != FUNC_RETURN_OK)
{
elog(WARNING, "Can not understand the value '%s' of property %s.",
rm_seg_memory_use,
HAWQDRM_CONFFILE_LIMIT_MEMORY_USE);
return MAIN_CONF_UNSET_SEGMENT_MEMORY_USE;
}
DRMGlobalInstance->SegmentCore = rm_seg_core_use;
elog(DEBUG3, "HAWQ RM :: Accepted NONE mode resource management setting, "
"each host has (%d MB,%lf) resource capacity.\n",
DRMGlobalInstance->SegmentMemoryMB,
DRMGlobalInstance->SegmentCore);
if ( DRMGlobalInstance->Role == START_RM_ROLE_SEGMENT )
{
// For temporary directories
InitTemporaryDirs(&DRMGlobalInstance->LocalHostTempDirectories, rm_seg_tmp_dirs);
DQUEUE_LOOP_BEGIN(&DRMGlobalInstance->LocalHostTempDirectories, iter, SimpStringPtr, value)
elog(LOG, "HAWQ Segment RM :: Temporary directory %s", value->Str);
DQUEUE_LOOP_END
checkAndBuildFailedTmpDirList();
}
/****** Resource enforcement GUCs begins ******/
/* Get resource enforcement enablement flag */
DRMGlobalInstance->ResourceEnforcerCpuEnable = rm_enforce_cpu_enable;
/* Get resource enforcement CGroup mount point */
initSimpleString(&(DRMGlobalInstance->ResourceEnforcerCgroupMountPoint),
PCONTEXT);
setSimpleStringNoLen(&(DRMGlobalInstance->ResourceEnforcerCgroupMountPoint),
rm_enforce_cgrp_mnt_pnt);
/* Get resource enforcement CGroup hierarchy name */
initSimpleString(&(DRMGlobalInstance->ResourceEnforcerCgroupHierarchyName),
PCONTEXT);
setSimpleStringNoLen(&(DRMGlobalInstance->ResourceEnforcerCgroupHierarchyName),
rm_enforce_cgrp_hier_name);
/* Get resource enforcement mapping between hawq weight and yarn weight */
DRMGlobalInstance->ResourceEnforcerCpuWeight = rm_enforce_cpu_weight;
/* Get resource enforcement vcore-pcore ratio */
DRMGlobalInstance->ResourceEnforcerVcorePcoreRatio = rm_enforce_core_vpratio;
/* Get resource enforcement CGroup cleanup period */
DRMGlobalInstance->ResourceEnforcerCleanupPeriod = rm_enforce_cleanup_period;
/****** Resource enforcement GUCs ends ******/
return FUNC_RETURN_OK;
}
/*
* Load resource queue and user information into HAWQ RM memory. If independent
* mode, the definition is from xml file saving a list of properties. If normal
* forked RM process, this can be loaded from catalog table: pg_authid and
* pg_resqueue.
*
*/
int loadAllQueueAndUser(void)
{
int res = FUNC_RETURN_OK;
List *queueprops = NULL; /* Save all queue properties. */
List *userprops = NULL; /* Save all role/user properties.*/
res = loadUserPropertiesFromCatalog(&userprops);
if ( res != FUNC_RETURN_OK )
{
goto exit;
}
res = loadQueuePropertiesFromCatalog(&queueprops);
if ( res != FUNC_RETURN_OK )
{
goto exit;
}
/* Add resource queue and user into resource pool data structure. */
res = addResourceQueueAndUserFromProperties(queueprops, userprops);
/* res is checked before returning. */
Assert(PQUEMGR->RootTrack != NULL);
Assert(PQUEMGR->DefaultTrack != NULL);
exit:
cleanPropertyList(PCONTEXT, &queueprops);
cleanPropertyList(PCONTEXT, &userprops);
if ( res != FUNC_RETURN_OK )
{
elog( LOG, "Fail to load queue and user definition.");
}
return res;
}
/*****************************************************************************
* Load user information from catalog table pg_authid.
*****************************************************************************/
int loadUserPropertiesFromCatalog(List **users)
{
int libpqres = CONNECTION_OK;
int ret = FUNC_RETURN_OK;
PGconn *conn = NULL;
static char conninfo[1024];
PQExpBuffer sql = NULL;
PGresult *result = NULL;
int ntups = 0;
int i_oid = 0,
i_rolname = 0,
i_rolsuper = 0,
i_rolresqueue = 0;
Oid roloid = 0,
rolresqueueOID = 0;
char *rolname = NULL,
*rolresqueue = NULL;
bool rolsuper = false;
int8_t DummyPriority = 3;
sprintf(conninfo, "options='-c gp_session_role=UTILITY' "
"dbname=template1 port=%d connect_timeout=%d",
master_addr_port,
LIBPQ_CONNECT_TIMEOUT);
conn = PQconnectdb(conninfo);
if ((libpqres = PQstatus(conn)) != CONNECTION_OK)
{
elog(WARNING, "Resource manager failed to connect database when loading"
"role specifications from pg_authid, error code %d, "
"reason: %s",
libpqres,
PQerrorMessage(conn));
PQfinish(conn);
return LIBPQ_FAIL_EXECUTE;
}
sql = createPQExpBuffer();
if ( sql == NULL )
{
elog(WARNING, "Resource manager failed to allocate buffer for building "
"sql statement.");
goto cleanup;
}
appendPQExpBuffer(sql, "SELECT oid,rolname,rolsuper,rolresqueue FROM pg_authid");
result = PQexec(conn, sql->data);
if (!result || PQresultStatus(result) != PGRES_TUPLES_OK)
{
elog(WARNING, "Resource manager failed to run SQL: %s "
"when loading role specifications from pg_authid, "
"reason : %s",
sql->data,
PQresultErrorMessage(result));
ret = LIBPQ_FAIL_EXECUTE;
goto cleanup;
}
ntups = PQntuples(result);
i_oid = PQfnumber(result, "oid");
i_rolname = PQfnumber(result, "rolname");
i_rolsuper = PQfnumber(result, "rolsuper");
i_rolresqueue = PQfnumber(result, "rolresqueue");
for (int i = 0; i < ntups; i++)
{
roloid = (Oid)strtoul(PQgetvalue(result, i, i_oid), NULL, 10);
rolname = PQgetvalue(result, i, i_rolname);
rolresqueue = PQgetvalue(result, i, i_rolresqueue);
if (rolresqueue == NULL || strlen(rolresqueue) == 0)
{
rolresqueueOID = InvalidOid;
}
else
{
rolresqueueOID = (Oid)strtoul(PQgetvalue(result, i, i_rolresqueue), NULL, 10);
}
if (PQgetvalue(result, i, i_rolsuper)[0] == 't')
{
rolsuper = true;
}
else
{
rolsuper = false;
}
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
*users = lappend(*users,
createPropertyOID(
PCONTEXT,
"user",
getUSRTBLAttributeName(USR_TBL_ATTR_OID),
&i,
roloid));
*users = lappend(*users,
createPropertyName(
PCONTEXT,
"user",
getUSRTBLAttributeName(USR_TBL_ATTR_NAME),
&i,
(Name)rolname));
*users = lappend(*users,
createPropertyOID(
PCONTEXT,
"user",
getUSRTBLAttributeName(USR_TBL_ATTR_TARGET_QUEUE),
&i,
rolresqueueOID));
*users = lappend(*users,
createPropertyBool(
PCONTEXT,
"user",
getUSRTBLAttributeName(USR_TBL_ATTR_IS_SUPERUSER),
&i,
rolsuper));
*users = lappend(*users,
createPropertyInt8(
PCONTEXT,
"user",
getUSRTBLAttributeName(USR_TBL_ATTR_PRIORITY),
&i,
DummyPriority));
MEMORY_CONTEXT_SWITCH_BACK
}
ListCell *cell = NULL;
foreach(cell, *users)
{
KVProperty property = lfirst(cell);
elog(RMLOG, "Resource manager loaded role specifications from pg_authid : "
"[%s]=[%s]",
property->Key.Str,
property->Val.Str);
}
elog(LOG, "Resource manager successfully loaded role specifications.");
cleanup:
if(sql != NULL)
{
destroyPQExpBuffer(sql);
}
if(result != NULL)
{
PQclear(result);
}
PQfinish(conn);
return ret;
}
/*****************************************************************************
* Load resource queue information from catalog table pg_resqueue.
*****************************************************************************/
int loadQueuePropertiesFromCatalog(List **queues)
{
int libpqres = CONNECTION_OK;
int ret = FUNC_RETURN_OK;
PGconn *conn = NULL;
static char conninfo[1024];
PQExpBuffer sql = NULL;
PGresult* result = NULL;
int ntups = 0;
int i_oid = 0,
i_name = 0,
i_parent = 0,
i_active_stats_cluster = 0,
i_memory_limit_cluster = 0,
i_core_limit_cluster = 0,
i_resource_overcommit = 0,
i_allocation_policy = 0,
i_vseg_resource_quota = 0,
i_nvseg_upper_limit = 0,
i_nvseg_lower_limit = 0,
i_nvseg_upper_limit_perseg = 0,
i_nvseg_lower_limit_perseg = 0,
i_creation_time = 0,
i_update_time = 0,
i_status = 0;
Oid oid = 0,
parentoid = 0;
char *name = NULL,
*parent = NULL,
*memory_limit_cluster = NULL,
*core_limit_cluster = NULL,
*allocation_policy = NULL,
*vseg_resource_quota = NULL,
*status = NULL;
int active_stats_cluster = 0,
nvseg_upper_limit = 0,
nvseg_lower_limit = 0;
float nvseg_upper_limit_perseg = 0.0,
nvseg_lower_limit_perseg = 0.0,
resource_overcommit = 0.0;
int64 creation_time = 0,
update_time = 0;
snprintf(conninfo, sizeof(conninfo),
"options='-c gp_session_role=UTILITY' "
"dbname=template1 port=%d connect_timeout=%d",
master_addr_port,
LIBPQ_CONNECT_TIMEOUT);
conn = PQconnectdb(conninfo);
if ((libpqres = PQstatus(conn)) != CONNECTION_OK)
{
elog(WARNING, "Resource manager failed to connect database when loading "
"resource queue specifications from pg_resqueue, "
"error code: %d, reason: %s",
libpqres,
PQerrorMessage(conn));
PQfinish(conn);
return LIBPQ_FAIL_EXECUTE;
}
sql = createPQExpBuffer();
if ( sql == NULL )
{
elog(WARNING, "Resource manager failed to allocate buffer for building "
"sql statement.");
goto cleanup;
}
appendPQExpBuffer(sql,"SELECT oid,"
"rsqname,"
"parentoid,"
"activestats,"
"memorylimit, "
"corelimit, "
"resovercommit,"
"allocpolicy, "
"vsegresourcequota, "
"nvsegupperlimit, "
"nvseglowerlimit, "
"nvsegupperlimitperseg, "
"nvseglowerlimitperseg, "
"creationtime, "
"updatetime, "
"status "
"FROM pg_resqueue");
result = PQexec(conn, sql->data);
if (!result || PQresultStatus(result) != PGRES_TUPLES_OK)
{
elog(WARNING, "Resource manager failed to run SQL: %s when loading "
"resource queue specifications from pg_resqueue, "
"reason : %s",
sql->data,
PQresultErrorMessage(result));
ret = LIBPQ_FAIL_EXECUTE;
goto cleanup;
}
ntups = PQntuples(result);
i_oid = PQfnumber(result, PG_RESQUEUE_COL_OID);
i_name = PQfnumber(result, PG_RESQUEUE_COL_RSQNAME);
i_parent = PQfnumber(result, PG_RESQUEUE_COL_PARENTOID);
i_active_stats_cluster = PQfnumber(result, PG_RESQUEUE_COL_ACTIVESTATS);
i_memory_limit_cluster = PQfnumber(result, PG_RESQUEUE_COL_MEMORYLIMIT);
i_core_limit_cluster = PQfnumber(result, PG_RESQUEUE_COL_CORELIMIT);
i_resource_overcommit = PQfnumber(result, PG_RESQUEUE_COL_RESOVERCOMMIT);
i_allocation_policy = PQfnumber(result, PG_RESQUEUE_COL_ALLOCPOLICY);
i_vseg_resource_quota = PQfnumber(result, PG_RESQUEUE_COL_VSEGRESOURCEQUOTA);
i_nvseg_upper_limit = PQfnumber(result, PG_RESQUEUE_COL_NVSEGUPPERLIMIT);
i_nvseg_lower_limit = PQfnumber(result, PG_RESQUEUE_COL_NVSEGLOWERLIMIT);
i_nvseg_upper_limit_perseg = PQfnumber(result, PG_RESQUEUE_COL_NVSEGUPPERLIMITPERSEG);
i_nvseg_lower_limit_perseg = PQfnumber(result, PG_RESQUEUE_COL_NVSEGLOWERLIMITPERSEG);
i_creation_time = PQfnumber(result, PG_RESQUEUE_COL_CREATIONTIME);
i_update_time = PQfnumber(result, PG_RESQUEUE_COL_UPDATETIME);
i_status = PQfnumber(result, PG_RESQUEUE_COL_STATUS);
for (int i = 0; i < ntups; i++)
{
oid = (Oid)strtoul(PQgetvalue(result, i, i_oid), NULL, 10);
name = PQgetvalue(result, i, i_name);
parent = PQgetvalue(result, i, i_parent);
if (parent == NULL || strlen(parent) == 0)
{
parentoid = InvalidOid;
}
else
{
parentoid = (Oid)strtoul(parent, NULL, 10);
}
parentoid = (Oid)strtoul(PQgetvalue(result, i, i_parent), NULL, 10);
active_stats_cluster = atoi(PQgetvalue(result, i, i_active_stats_cluster));
memory_limit_cluster = PQgetvalue(result, i, i_memory_limit_cluster);
core_limit_cluster = PQgetvalue(result, i, i_core_limit_cluster);
resource_overcommit = atof(PQgetvalue(result, i, i_resource_overcommit));
allocation_policy = PQgetvalue(result, i, i_allocation_policy);
vseg_resource_quota = PQgetvalue(result, i, i_vseg_resource_quota);
nvseg_upper_limit = atoi(PQgetvalue(result, i, i_nvseg_upper_limit));
nvseg_lower_limit = atoi(PQgetvalue(result, i, i_nvseg_lower_limit));
nvseg_upper_limit_perseg = atof(PQgetvalue(result, i, i_nvseg_upper_limit_perseg));
nvseg_lower_limit_perseg = atof(PQgetvalue(result, i, i_nvseg_lower_limit_perseg));
creation_time = atol(PQgetvalue(result, i, i_creation_time));
update_time = atol(PQgetvalue(result, i, i_update_time));
status = PQgetvalue(result, i, i_status);
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
*queues = lappend(*queues,
createPropertyOID(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_OID),
&i,
oid));
*queues = lappend(*queues,
createPropertyName(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_NAME),
&i,
(Name)name));
*queues = lappend(*queues,
createPropertyOID(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_PARENT),
&i,
parentoid));
*queues = lappend(*queues,
createPropertyInt32(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_ACTIVE_STATMENTS),
&i,
active_stats_cluster));
*queues = lappend(*queues,
createPropertyString(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_MEMORY_LIMIT_CLUSTER),
&i,
memory_limit_cluster));
*queues = lappend(*queues,
createPropertyString(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_CORE_LIMIT_CLUSTER),
&i,
core_limit_cluster));
*queues = lappend(*queues,
createPropertyString(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_ALLOCATION_POLICY),
&i,
allocation_policy));
*queues = lappend(*queues,
createPropertyFloat(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_RESOURCE_OVERCOMMIT_FACTOR),
&i,
resource_overcommit));
*queues = lappend(*queues,
createPropertyString(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_VSEG_RESOURCE_QUOTA),
&i,
vseg_resource_quota));
*queues = lappend(*queues,
createPropertyInt32(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_NVSEG_UPPER_LIMIT),
&i,
nvseg_upper_limit));
*queues = lappend(*queues,
createPropertyInt32(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_NVSEG_LOWER_LIMIT),
&i,
nvseg_lower_limit));
*queues = lappend(*queues,
createPropertyFloat(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_NVSEG_UPPER_LIMIT_PERSEG),
&i,
nvseg_upper_limit_perseg));
*queues = lappend(*queues,
createPropertyFloat(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_NVSEG_LOWER_LIMIT_PERSEG),
&i,
nvseg_lower_limit_perseg));
*queues = lappend(*queues,
createPropertyInt32(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_CREATION_TIME),
&i,
creation_time));
*queues = lappend(*queues,
createPropertyInt32(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_UPDATE_TIME),
&i,
update_time));
*queues = lappend(*queues,
createPropertyString(
PCONTEXT,
"queue",
getRSQTBLAttributeName(RSQ_TBL_ATTR_STATUS),
&i,
status));
MEMORY_CONTEXT_SWITCH_BACK
}
ListCell *cell = NULL;
foreach(cell, *queues)
{
KVProperty property = lfirst(cell);
elog(RMLOG, "Resource manger loaded resource queue specifications from "
"pg_resqueue : [%s]=[%s]",
property->Key.Str,
property->Val.Str);
}
elog(LOG, "Resource manger successfully loaded resource queue specifications");
cleanup:
if(sql != NULL)
{
destroyPQExpBuffer(sql);
}
if(result != NULL)
{
PQclear(result);
}
PQfinish(conn);
return ret;
}
void quitResManager(SIGNAL_ARGS)
{
DRMGlobalInstance->ResManagerMainKeepRun = false;
}
void handleChildSignal(SIGNAL_ARGS)
{
/* The resource broker process quits. */
RB_handleSignalSIGCHLD();
}
void notifyPostmasterResManagerStarted(SIGNAL_ARGS)
{
DRMStartup = true;
}
/*
* Read properties and generate corresponding queue and user definitions.
*
* properties[in] : list of KVPropertyData
* queues[out] : list of DynResourceQueue
* users[out] : list of UserInfo
*/
int addResourceQueueAndUserFromProperties(List *queueprops, List *userprops)
{
static char errorbuf[1024];
int res = FUNC_RETURN_OK;
char *substrstart = NULL;
int substrlen = 0;
char *substr2start = NULL;
int substr2len = 0;
List *queueattrlist = NULL;
List *userattrlist = NULL;
ListCell *cell = NULL;
/*
* STEP 1. Process properties and build resource queue and user structure.
*/
int currentindex = -1;
List *currentattrs = NULL;
foreach(cell, queueprops)
{
KVProperty value = lfirst(cell);
elog(RMLOG, "Loads queue property %s=%s", value->Key.Str, value->Val.Str);
/* Split key string into (attribute, index) */
if ( SimpleStringStartWith(&(value->Key), "queue.") != FUNC_RETURN_OK )
{
elog(RMLOG, "Ignore property %s=%s", value->Key.Str, value->Val.Str);
continue;
}
/*
*----------------------------------------------------------------------
* The key string is formatted as
* queue.<queueattrstr>.<queueindexstr>=<value>
* (0) (1) (2)
*----------------------------------------------------------------------
*/
int32_t queueindex;
SimpString queueattrstr;
SimpString queueindexstr;
/* Fill queueindex, queueattrstr */
int parseres1 = PropertyKeySubstring(&(value->Key),
1,
&substrstart,
&substrlen);
int parseres2 = PropertyKeySubstring(&(value->Key),
2,
&substr2start,
&substr2len);
if( parseres1 != FUNC_RETURN_OK || parseres2 != FUNC_RETURN_OK )
{
Assert(false);
}
setSimpleStringRef(&queueindexstr, substr2start, substr2len);
setSimpleStringRef(&queueattrstr, substrstart, substrlen);
parseres1 = SimpleStringToInt32(&queueindexstr, &queueindex);
if ( parseres1 != FUNC_RETURN_OK )
{
Assert(false);
}
/* Fill queue attribute to corresponding list. */
KVProperty newprop = createPropertyEmpty(PCONTEXT);
setSimpleStringWithContent(&(newprop->Key), substrstart, substrlen);
setSimpleStringNoLen(&(newprop->Val), value->Val.Str);
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
if ( queueindex != currentindex )
{
if ( currentattrs != NULL )
{
queueattrlist = lappend(queueattrlist, currentattrs);
}
currentattrs = NULL;
currentindex = queueindex;
}
elog(RMLOG, "Resource manager loaded attribute for creating queue %s=%s",
newprop->Key.Str,
newprop->Val.Str);
currentattrs = lappend(currentattrs, newprop);
MEMORY_CONTEXT_SWITCH_BACK
}
if ( currentattrs != NULL )
{
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
queueattrlist = lappend(queueattrlist, currentattrs);
MEMORY_CONTEXT_SWITCH_BACK
}
currentindex = -1;
currentattrs = NULL;
foreach(cell, userprops)
{
KVProperty value = lfirst(cell);
elog(RMLOG, "Loads user property %s=%s", value->Key.Str, value->Val.Str);
/* Split key string into (attribute, index) */
if ( SimpleStringStartWith(&(value->Key), "user.") != FUNC_RETURN_OK )
{
elog(RMLOG, "Ignore property %s=%s", value->Key.Str, value->Val.Str);
continue;
}
/*
*----------------------------------------------------------------------
* The key string is formatted as
* user.<userattrstr>.<userindexstr>=<value>
* (0) (1) (2)
*----------------------------------------------------------------------
*/
int32_t userindex;
SimpString userattrstr;
SimpString userindexstr;
/* Fill userindex, userattrstr */
int parseres1 = PropertyKeySubstring(&(value->Key),
1,
&substrstart,
&substrlen);
int parseres2 = PropertyKeySubstring(&(value->Key),
2,
&substr2start,
&substr2len);
if ( parseres1 != FUNC_RETURN_OK || parseres2 != FUNC_RETURN_OK )
{
Assert(false);
}
setSimpleStringRef(&userindexstr, substr2start, substr2len);
setSimpleStringRef(&userattrstr, substrstart, substrlen);
parseres1 = SimpleStringToInt32(&userindexstr, &userindex);
Assert(parseres1 == FUNC_RETURN_OK);
/* Fill user attribute to corresponding list. */
KVProperty newprop = createPropertyEmpty(PCONTEXT);
setSimpleStringWithContent(&(newprop->Key), substrstart, substrlen);
setSimpleStringNoLen(&(newprop->Val), value->Val.Str);
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
if ( userindex != currentindex )
{
if ( currentattrs != NULL )
{
userattrlist = lappend(userattrlist, currentattrs);
}
currentattrs = NULL;
currentindex = userindex;
}
elog(RMLOG, "Resource manager loaded attribute for creating role %s=%s",
newprop->Key.Str,
newprop->Val.Str);
currentattrs = lappend(currentattrs, newprop);
MEMORY_CONTEXT_SWITCH_BACK
}
if ( currentattrs != NULL )
{
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
userattrlist = lappend(userattrlist, currentattrs);
MEMORY_CONTEXT_SWITCH_BACK
}
/*
* STEP 2. Add resource queues.
*/
/* STEP 2.1. Build resource queue object with parsed properties set. */
List *rawrsqs = NULL;
foreach(cell, queueattrlist)
{
List *attrs = lfirst(cell);
ListCell *cell2 = NULL;
foreach(cell2, attrs)
{
KVProperty attrkv = lfirst(cell2);
elog(RMLOG, "To parse : %s=%s", attrkv->Key.Str, attrkv->Val.Str);
}
DynResourceQueue newqueue = rm_palloc0(PCONTEXT,
sizeof(DynResourceQueueData));
res = parseResourceQueueAttributes(attrs,
newqueue,
false,
true,
errorbuf,
sizeof(errorbuf));
if ( res != FUNC_RETURN_OK )
{
rm_pfree(PCONTEXT, newqueue);
elog(WARNING, "Resource manager can not create resource queue with its "
"attributes because %s",
errorbuf);
continue;
}
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
rawrsqs = lappend(rawrsqs, newqueue);
MEMORY_CONTEXT_SWITCH_BACK
}
/*
* STEP 2.2. Reorder the resource queue sequence to ensure that every time
* the queue is added into the manager, its parent queue can be found.
*/
List *orderedrsqs = NULL;
bool orderchanged = true;
while( orderchanged )
{
DynResourceQueue toreordrsq = NULL;
ListCell *cell = NULL;
ListCell *prevcell = NULL;
foreach(cell, rawrsqs)
{
DynResourceQueue rawqueue = lfirst(cell);
if ( rawqueue->ParentOID == InvalidOid )
{
toreordrsq = rawqueue;
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
rawrsqs = list_delete_cell(rawrsqs, cell, prevcell);
MEMORY_CONTEXT_SWITCH_BACK
break;
}
ListCell *cell2 = NULL;
foreach(cell2, orderedrsqs)
{
DynResourceQueue ordqueue = lfirst(cell2);
if ( ordqueue->OID == rawqueue->ParentOID )
{
toreordrsq = rawqueue;
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
rawrsqs = list_delete_cell(rawrsqs, cell, prevcell);
MEMORY_CONTEXT_SWITCH_BACK
}
}
if ( toreordrsq != NULL )
{
break;
}
prevcell = cell;
}
orderchanged = false;
if ( toreordrsq != NULL )
{
elog(RMLOG, "Find one resource queue valid to continue loading %s.",
toreordrsq->Name);
orderchanged = true;
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
orderedrsqs = lappend(orderedrsqs, toreordrsq);
MEMORY_CONTEXT_SWITCH_BACK
}
}
/*
* There is one possibility that we have some resource queues whose parent
* queue ids are not present in the pg_resqueue table or external configure
* file. In this case, these resource queues are logged out out as warnings,
* and HAWQ RM will ignore them.
*
* Through catalog, this should be impossible.
*/
if ( list_length(rawrsqs) > 0 )
{
elog(WARNING, "Invalid resource queues are detected due to no valid "
"parent resource queue id.");
while( list_length(rawrsqs) > 0 )
{
DynResourceQueue rawqueue = lfirst(list_head(rawrsqs));
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
rawrsqs = list_delete_first(rawrsqs);
MEMORY_CONTEXT_SWITCH_BACK
elog(WARNING, "Invalid resource queue : [%s]", rawqueue->Name);
rm_pfree(PCONTEXT, rawqueue);
}
}
/*
* STEP 2.3. List of DynResourceQueue, save partially filled attributes.
* Actually check and complete the resource queue definition and save it
* in the resource queue manager.
*/
while( list_length(orderedrsqs) > 0 )
{
DynResourceQueue partqueue = lfirst(list_head(orderedrsqs));
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
orderedrsqs = list_delete_first(orderedrsqs);
MEMORY_CONTEXT_SWITCH_BACK
elog(RMLOG, "Load queue %s.", partqueue->Name);
res = checkAndCompleteNewResourceQueueAttributes(partqueue,
errorbuf,
sizeof(errorbuf));
if ( res != FUNC_RETURN_OK )
{
elog(RMLOG, "res=%d error=%s, after check and complete queue %s.",
res,
errorbuf,
partqueue->Name);
rm_pfree(PCONTEXT, partqueue);
elog(WARNING, "Resource manager can not complete resource queue's "
"attributes because %s",
errorbuf);
continue;
}
elog(RMLOG, "Checked and completed queue %s.", partqueue->Name);
DynResourceQueueTrack newtrack = NULL;
res = createQueueAndTrack(partqueue, &newtrack, errorbuf, sizeof(errorbuf));
if ( res != FUNC_RETURN_OK )
{
rm_pfree(PCONTEXT, partqueue);
if ( newtrack != NULL )
{
rm_pfree(PCONTEXT, newtrack);
}
elog( WARNING, "Resource manager can not create resource queue %s "
"because %s",
partqueue->Name,
errorbuf);
continue;
}
elog(RMLOG, "Created queue %s.", partqueue->Name);
char buffer[1024];
generateQueueReport(partqueue->OID, buffer, sizeof(buffer));
elog(LOG, "Resource manager created resource queue instance : %s",
buffer);
}
/*
* STEP 3. Add users.
*/
while( list_length(userattrlist) > 0 )
{
List *attrs = lfirst(list_head(userattrlist));
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
userattrlist = list_delete_first(userattrlist);
MEMORY_CONTEXT_SWITCH_BACK
UserInfo newuser = rm_palloc0(PCONTEXT,
sizeof(UserInfoData));
res = parseUserAttributes(attrs,newuser, errorbuf, sizeof(errorbuf));
if ( res != FUNC_RETURN_OK )
{
elog(WARNING, "Can not create user with its attributes because %s",
errorbuf);
rm_pfree(PCONTEXT, newuser);
continue;
}
res = checkUserAttributes(newuser, errorbuf, sizeof(errorbuf));
if ( res != FUNC_RETURN_OK )
{
elog(WARNING, "User is not valid because %s", errorbuf);
rm_pfree(PCONTEXT, newuser);
continue;
}
createUser(newuser);
char buffer[256];
generateUserReport(newuser->Name,
strlen(newuser->Name),
buffer,
sizeof(buffer));
elog(LOG, "Resource manager created user. %s", buffer);
}
return res;
}
int generateAllocRequestToBroker(void)
{
int res = FUNC_RETURN_OK;
/*--------------------------------------------------------------------------
* This is a temporary restrict that HAWQ RM supports only one memory/core
* ratio in current version.
*--------------------------------------------------------------------------
*/
Assert( PQUEMGR->RatioCount == 1 );
DynMemoryCoreRatioTrack mctrack = PQUEMGR->RatioTrackers[0];
bool hasWorkload = mctrack->TotalUsed.MemoryMB +
mctrack->TotalRequest.MemoryMB > 0;
if ( !hasWorkload )
{
/* Check if resource manager has workload recently. */
if ( PQUEMGR->RatioCount == 1 &&
PQUEMGR->RatioWaterMarks[0].NodeCount > 0 )
{
DynMemoryCoreRatioWaterMark mark =
(DynMemoryCoreRatioWaterMark)
getDQueueHeadNodeData(&(PQUEMGR->RatioWaterMarks[0]));
hasWorkload = mark->ClusterMemoryMB > 0 ;
}
}
/* Decide water level of resource. */
int wlevel = hasWorkload ? PQUEMGR->ActualMinGRMContainerPerSeg : 0;
switch( DRMGlobalInstance->ImpType )
{
case YARN_LIBYARN:
/* Do nothing. */
break;
case NONE_HAWQ2:
/* We always expect all resource allocated. */
wlevel = INT_MAX;
break;
default:
Assert(false);
}
/*
* Go through each segment, decide how many containers should be allocated.
* And generate preferred host list based on minimum water level.
*/
int grmctnsize = 0;
List *preferred = NULL;
List *ressegl = NULL;
ListCell *cell = NULL;
getAllPAIRRefIntoList(&(PRESPOOL->Segments), &ressegl);
foreach(cell, ressegl)
{
PAIR pair = (PAIR)lfirst(cell);
SegResource segres = (SegResource)(pair->Value);
/*
* Resource manager skips this segment if
* 1) Not FTS available;
* 2) Having resource decrease pending.
*/
if (!IS_SEGSTAT_FTSAVAILABLE(segres->Stat) ||
(segres->DecPending.MemoryMB > 0 && segres->DecPending.Core > 0))
{
continue;
}
/* Get segment resource capacity. */
uint32_t memcap = getSegResourceCapacityMemory(segres);
uint32_t corecap = getSegResourceCapacityCore(segres);
/*
* The segment capacity mem/core ratio maybe not equal to the cluster
* mem/core ratio.
*/
int ctnsize = memcap / mctrack->MemCoreRatio;
ctnsize = ctnsize < corecap ? ctnsize : corecap;
/* Follow the water level calculated. */
ctnsize = ctnsize < wlevel ? ctnsize : wlevel;
elog(RMLOG, "Host %s has %d GRM containers, (%d MB, %lf CORE) increase pending.",
GET_SEGRESOURCE_HOSTNAME(segres),
(segres->ContainerSets[0] == NULL ?
0 :
list_length(segres->ContainerSets[0]->Containers)),
segres->IncPending.MemoryMB,
segres->IncPending.Core);
int ctnsizeneed = ctnsize -
(segres->ContainerSets[0] == NULL ?
0 :
list_length(segres->ContainerSets[0]->Containers)) -
segres->IncPending.MemoryMB / mctrack->MemCoreRatio;
if ( ctnsizeneed <= 0 )
{
/*
* If the segment contains more resource than expected resource
* water level.
*/
continue;
}
/* Build preferred host information. */
PAIR newpair = rm_palloc0(PCONTEXT, sizeof(PAIRData));
newpair->Key = segres;
ResourceBundle resource = rm_palloc0(PCONTEXT, sizeof(ResourceBundleData));
resource->MemoryMB = mctrack->MemCoreRatio * (ctnsizeneed);
resource->Core = ctnsizeneed;
newpair->Value = resource;
preferred = lappend(preferred, newpair);
elog(DEBUG3, "Expect resource from resource broker with "
"preferred host, hostname:%s, container number:%lf.",
GET_SEGRESOURCE_HOSTNAME(segres),
resource->Core);
grmctnsize += ctnsizeneed;
}
freePAIRRefList(&(PRESPOOL->Segments), &ressegl);
elog(RMLOG, "Resource manager needs minimum %d GRM containers.", grmctnsize);
/* Decide how much resource to acquire from global resource manager. */
int32_t reqmem = 0;
int32_t reqcore = ceil(mctrack->TotalRequest.Core +
mctrack->TotalUsed.Core -
mctrack->TotalAllocated.Core -
mctrack->TotalPending.Core);
elog(RMLOG, "Memory Core Track has %lf requested, %lf used, %lf allocated, "
"%lf pending.",
mctrack->TotalRequest.Core,
mctrack->TotalUsed.Core,
mctrack->TotalAllocated.Core,
mctrack->TotalPending.Core);
/*
* If after checking water level of segment resource, we expect more resource
* containers, we expect more as well.
*/
reqcore = reqcore < grmctnsize - mctrack->TotalPending.Core ?
grmctnsize - mctrack->TotalPending.Core :
reqcore;
reqmem = reqcore * mctrack->MemCoreRatio;
elog(RMLOG, "Resource manager now needs %d GRM containers.", reqcore);
/*
* Check if should raise water level to deal with resource fragment or
* resource uneven problems. We trigger this logic only when no resource
* request caused by lack of resource, and no pending resource are waited
* for.
*/
if ( reqcore <= 0 &&
mctrack->TotalPending.Core <= 0 &&
(PQUEMGR->hasResourceProblem[RESPROBLEM_FRAGMENT] ||
PQUEMGR->hasResourceProblem[RESPROBLEM_UNEVEN] ||
PQUEMGR->hasResourceProblem[RESPROBLEM_TOOFEWSEG]) )
{
/* Check if it is possible to raise water level. */
if ( mctrack->TotalAllocated.Core + 1 <=
PRESPOOL->GRMTotal.Core * PQUEMGR->GRMQueueMaxCapacity )
{
/*
* We only add one more GRM container to acquire, this will trigger
* the following logic to raise the water level.
*/
reqcore = 1;
reqmem = reqcore * mctrack->MemCoreRatio;
PQUEMGR->hasResourceProblem[RESPROBLEM_FRAGMENT] = false;
PQUEMGR->hasResourceProblem[RESPROBLEM_UNEVEN] = false;
PQUEMGR->hasResourceProblem[RESPROBLEM_TOOFEWSEG] = false;
elog(LOG, "Resource manager raises segment resource water level.");
}
}
/* Call resource broker to request resource. */
if ( reqmem > 0 && reqcore > 0 )
{
/*
* Here we know that we have to allocate more resource from GRM, we should
* check again to enrich the preferred host list for new locality data
* of expected additional GRM containers.
*
* The expected size of GRM containers may vary because we expect GRM
* to have even number of GRM containers in each available segments.
*/
completeAllocRequestToBroker(&reqmem, &reqcore, &preferred);
elog(RMLOG, "Resource manager needs %d GRM containers after adjusting "
"overall water level.",
reqcore);
if ( reqcore > 0 )
{
PRESPOOL->LastResAcqTime = gettime_microsec();
addResourceBundleData(&(mctrack->TotalPending), reqmem, reqcore);
uint64_t oldtime = mctrack->TotalPendingStartTime;
if ( mctrack->TotalPendingStartTime == 0 )
{
mctrack->TotalPendingStartTime = gettime_microsec();
elog(DEBUG3, "Global resource total pending start time is updated "
"to "UINT64_FORMAT,
mctrack->TotalPendingStartTime);
}
/* Generate request to resource broker now. */
res = RB_acquireResource(reqmem, reqcore, preferred);
if ( res != FUNC_RETURN_OK && res != RESBROK_PIPE_BUSY )
{
minusResourceBundleData(&(mctrack->TotalPending), reqmem, reqcore);
mctrack->TotalPendingStartTime = oldtime;
elog(WARNING, "Resource manager failed to allocate resource from "
"resource broker (%d MB, %d CORE).",
reqmem, reqcore);
}
else if ( res == RESBROK_PIPE_BUSY )
{
minusResourceBundleData(&(mctrack->TotalPending), reqmem, reqcore);
elog(DEBUG3, "Resource manager should retry to submit request.");
res = FUNC_RETURN_OK;
}
else
{
elog(RMLOG, "Resource manager finished submitting resource "
"allocation request to global resource manager for "
"(%d MB, %d CORE).",
reqmem,
reqcore);
}
}
}
return res;
}
void completeAllocRequestToBroker(int32_t *reqmem,
int32_t *reqcore,
List **preferred)
{
/*
* Go through each segment to get minimum water level. The idea of completing
* the request is to keep pulling up the lowest water level in the cluster
* until equal or more GRM containers are requested.
*/
Assert(*reqmem % *reqcore == 0);
uint32_t ratio = *reqmem / *reqcore;
/* Step 1. Get lowest water level and build up index. */
List *ressegl = NULL;
getAllPAIRRefIntoList(&(PRESPOOL->Segments), &ressegl);
/* Index of each segment in current preferred host list. */
PAIR *reqidx = rm_palloc0(PCONTEXT, sizeof(PAIR) * list_length(ressegl));
int llevel = INT_MAX;
int totalcount = 0;
int index = 0;
bool allunavail = true;
ListCell *cell = NULL;
foreach(cell, ressegl)
{
reqidx[index] = NULL;
PAIR pair = (PAIR)lfirst(cell);
SegResource segres = (SegResource)(pair->Value);
/*
* Resource manager skips this segment if
* 1) Not FTS available;
* 2) Having resource decrease pending.
*/
if (!IS_SEGSTAT_FTSAVAILABLE(segres->Stat) ||
(segres->DecPending.MemoryMB > 0 && segres->DecPending.Core > 0))
{
index++;
continue;
}
allunavail = false;
int clevel = (segres->ContainerSets[0] == NULL ?
0 :
list_length(segres->ContainerSets[0]->Containers)) +
segres->IncPending.MemoryMB / ratio;
ListCell *pcell = NULL;
foreach(pcell, *preferred)
{
PAIR existpair = (PAIR)lfirst(pcell);
if ( existpair->Key == segres )
{
reqidx[index] = existpair;
totalcount += ((ResourceBundle)(reqidx[index]->Value))->MemoryMB /
ratio;
break;
}
}
int creqsize = reqidx[index] == NULL ?
0 :
((ResourceBundle)(reqidx[index]->Value))->MemoryMB / ratio;
llevel = (clevel+creqsize) < llevel ? (clevel+creqsize) : llevel;
index++;
}
/* If no segment available for adjusting the request, no need to keep going */
if ( allunavail )
{
rm_pfree(PCONTEXT, reqidx);
return;
}
for ( int i = 0 ; i < list_length(ressegl) ; ++i )
{
if ( reqidx[i] == NULL )
{
continue;
}
elog(RMLOG, "Expect resource from resource broker with "
"preferred host, hostname:%s, container number:%lf, "
"increase pending %d.",
GET_SEGRESOURCE_HOSTNAME(((SegResource)(reqidx[i]->Key))),
((ResourceBundle)(reqidx[i]->Value))->Core,
((SegResource)(reqidx[i]->Key))->IncPending.MemoryMB/ratio);
}
elog(RMLOG, "Lowest water level before adjusting is %d.", llevel);
/* Step 2. Adjust request. */
int32_t reqcoreleft = *reqcore - totalcount;
bool keeplooping = true;
while( reqcoreleft > 0 && keeplooping )
{
llevel++;
index = 0;
keeplooping = false;
foreach(cell, ressegl)
{
PAIR pair = (PAIR)lfirst(cell);
SegResource segres = (SegResource)(pair->Value);
/*
* Resource manager skips this segment if
* 1) Not FTS available;
* 2) Having resource decrease pending.
*/
if (!IS_SEGSTAT_FTSAVAILABLE(segres->Stat) ||
(segres->DecPending.MemoryMB > 0 && segres->DecPending.Core > 0))
{
index++;
continue;
}
/* Check total capacity of the segment. */
uint32_t corecap = getSegResourceCapacityCore(segres);
int clevel = (segres->ContainerSets[0] == NULL ?
0 :
list_length(segres->ContainerSets[0]->Containers)) +
segres->IncPending.MemoryMB / ratio;
int aclevel = reqidx[index] == NULL ?
clevel :
(clevel + ((ResourceBundle)(reqidx[index]->Value))->MemoryMB / ratio);
if ( llevel > aclevel && llevel <= corecap )
{
if ( reqidx[index] == NULL )
{
reqidx[index] = rm_palloc0(PCONTEXT, sizeof(PAIRData));
reqidx[index]->Key = segres;
ResourceBundle resource = rm_palloc0(PCONTEXT,
sizeof(ResourceBundleData));
resetResourceBundleData(resource, 0, 0, ratio);
reqidx[index]->Value = resource;
*preferred = lappend(*preferred, reqidx[index]);
}
addResourceBundleData((ResourceBundle)(reqidx[index]->Value),
(llevel-aclevel) * ratio,
llevel-aclevel);
reqcoreleft -= llevel-aclevel;
keeplooping = true;
elog(RMLOG, "Resource manager acquires %lf GRM containers on "
"host %s. Current level(having pending) %d, "
"expect level %d, acquired in current request %d.",
((ResourceBundle)(reqidx[index]->Value))->Core,
GET_SEGRESOURCE_HOSTNAME(((SegResource)(reqidx[index]->Key))),
clevel,
llevel,
aclevel-clevel);
}
index++;
}
}
/* Adjust total mem and core request. */
*reqcore -= reqcoreleft;
*reqmem = *reqcore * ratio;
rm_pfree(PCONTEXT, reqidx);
}
/*
* Create all socket servers to accept connection from QD or RM agents.
*/
int initializeSocketServer(void)
{
int res = FUNC_RETURN_OK;
int netres = 0;
char *allip = "0.0.0.0";
pgsocket RMListenSocket[HAWQRM_SERVER_PORT_COUNT];
for ( int i = 0 ; i < HAWQRM_SERVER_PORT_COUNT ; ++i )
{
RMListenSocket[i] = PGINVALID_SOCKET;
}
netres = StreamServerPort(AF_UNSPEC,
allip,
rm_master_port,
NULL,
RMListenSocket,
HAWQRM_SERVER_PORT_COUNT);
if ( netres != STATUS_OK )
{
res = REQUESTHANDLER_FAIL_START_SOCKET_SERVER;
elog(LOG, "Resource manager cannot create socket server. Port=%d",
rm_master_port);
return res;
}
/* Initialize array for polling all file descriptors. */
initializeAsyncComm();
int validfdcount = 0;
AsyncCommBuffer newbuffer = NULL;
for ( int i = 0 ; i < HAWQRM_SERVER_PORT_COUNT ; ++i )
{
if (RMListenSocket[i] != PGINVALID_SOCKET)
{
netres = registerFileDesc(RMListenSocket[i],
ASYNCCOMM_READ,
&AsyncCommBufferHandlersMsgServer,
NULL,
&newbuffer);
if ( netres != FUNC_RETURN_OK )
{
res = REQUESTHANDLER_FAIL_START_SOCKET_SERVER;
elog(WARNING, "Resource manager cannot track socket server.");
break;
}
validfdcount++;
InitHandler_Message(newbuffer);
}
}
if ( res != FUNC_RETURN_OK )
{
for ( int i = 0 ; i < HAWQRM_SERVER_PORT_COUNT ; ++i )
{
if ( RMListenSocket[i] != PGINVALID_SOCKET )
{
close(RMListenSocket[i]);
}
}
return res;
}
elog(LOG, "Resource manager starts accepting resource request. "
"Listening normal socket port %d. "
"Total listened %d FDs.",
rm_master_port,
validfdcount);
return res;
}
/*
* Send built response content back to RM. The sending is asynchronous.
*
* If the progress is error, close the connection.
*/
void sendResponseToClients(void)
{
while( list_length(PCONTRACK->ConnToSend) > 0 )
{
ConnectionTrack conntrack = (ConnectionTrack)
lfirst(list_head(PCONTRACK->ConnToSend));
MEMORY_CONTEXT_SWITCH_TO(PCONTEXT)
PCONTRACK->ConnToSend = list_delete_first(PCONTRACK->ConnToSend);
MEMORY_CONTEXT_SWITCH_BACK
if ( conntrack->CommBuffer != NULL )
{
buildMessageToCommBuffer(conntrack->CommBuffer,
conntrack->MessageBuff.Buffer,
conntrack->MessageSize,
conntrack->MessageID,
conntrack->MessageMark1,
conntrack->MessageMark2);
}
}
}
/*
* Check and set the nodes down that are not updated by IMAlive heart-beat for a
* long time.
*/
void updateStatusOfAllNodes()
{
SegResource node = NULL;
uint64_t curtime = 0;
bool changedstatus = false;
curtime = gettime_microsec();
for(uint32_t idx = 0; idx < PRESPOOL->SegmentIDCounter; idx++)
{
node = getSegResource(idx);
Assert(node != NULL);
uint8_t oldStatus = node->Stat->FTSAvailable;
if ( (curtime - node->LastUpdateTime >
1000000LL * rm_segment_heartbeat_timeout) &&
(node->Stat->StatusDesc & SEG_STATUS_HEARTBEAT_TIMEOUT) == 0)
{
/*
* This segment is heartbeat timeout, update its description
* and set it to unavailable if needed.
*/
if (oldStatus == RESOURCE_SEG_STATUS_AVAILABLE)
{
/*
* This call makes resource manager able to adjust queue and mem/core
* trackers' capacity.
*/
setSegResHAWQAvailability(node, RESOURCE_SEG_STATUS_UNAVAILABLE);
/*
* This call makes resource pool remove unused containers.
*/
returnAllGRMResourceFromSegment(node);
changedstatus = true;
}
node->Stat->StatusDesc |= SEG_STATUS_HEARTBEAT_TIMEOUT;
if (Gp_role != GP_ROLE_UTILITY)
{
SimpStringPtr description = build_segment_status_description(node->Stat);
update_segment_status(idx + REGISTRATION_ORDER_OFFSET,
SEGMENT_STATUS_DOWN,
(description->Len > 0)?description->Str:"");
add_segment_history_row(idx + REGISTRATION_ORDER_OFFSET,
GET_SEGRESOURCE_HOSTNAME(node),
(description->Len > 0)?description->Str:"");
freeSimpleStringContent(description);
rm_pfree(PCONTEXT, description);
}
elog(WARNING, "Resource manager sets host %s heartbeat timeout.",
GET_SEGRESOURCE_HOSTNAME(node));
}
}
if ( changedstatus )
{
refreshResourceQueueCapacity(false);
refreshActualMinGRMContainerPerSeg();
}
validateResourcePoolStatus(true);
}
/*
* Set all nodes DOWN due to no GRM cluster report
*/
void setAllNodesGRMDown()
{
SegResource node = NULL;
bool changedstatus = false;
for(uint32_t idx = 0; idx < PRESPOOL->SegmentIDCounter; idx++)
{
node = getSegResource(idx);
Assert(node != NULL);
uint8_t oldStatus = node->Stat->FTSAvailable;
uint32_t oldDesc = node->Stat->StatusDesc;
if (oldStatus == RESOURCE_SEG_STATUS_AVAILABLE)
{
/*
* This call makes resource manager able to adjust queue and mem/core
* trackers' capacity.
*/
setSegResHAWQAvailability(node, RESOURCE_SEG_STATUS_UNAVAILABLE);
changedstatus = true;
}
node->Stat->StatusDesc |= SEG_STATUS_NO_GRM_NODE_REPORT;
if (Gp_role != GP_ROLE_UTILITY && oldDesc != node->Stat->StatusDesc)
{
SimpStringPtr description = build_segment_status_description(node->Stat);
update_segment_status(idx + REGISTRATION_ORDER_OFFSET,
SEGMENT_STATUS_DOWN,
(description->Len > 0)?description->Str:"");
add_segment_history_row(idx + REGISTRATION_ORDER_OFFSET,
GET_SEGRESOURCE_HOSTNAME(node),
(description->Len > 0)?description->Str:"");
freeSimpleStringContent(description);
rm_pfree(PCONTEXT, description);
}
if (oldDesc != node->Stat->StatusDesc)
{
elog(WARNING, "Resource manager sets host %s down in cleanup phase "
"for resource broker error.",
GET_SEGRESOURCE_HOSTNAME(node));
}
}
if (changedstatus)
{
refreshResourceQueueCapacity(false);
refreshActualMinGRMContainerPerSeg();
}
}
/**
* Load segment information from file.
*
* NOTE: This is a test facility.
*/
int loadHostInformationIntoResourcePool(void)
{
int res = FUNC_RETURN_OK;
SelfMaintainBufferData seginfobuff;
FILE *phostlist = NULL;
/* Open the file */
elog(LOG, "HAWQ RM :: To load file %s", rm_resourcepool_test_filename);
phostlist = fopen(rm_resourcepool_test_filename, "r");
if ( phostlist == NULL )
return FUNC_RETURN_OK;
/* This buffer is used for building machine id instance. */
initializeSelfMaintainBuffer(&seginfobuff, PCONTEXT);
/*
* Loop each host, we identify hosts by hostname only. The first segment in
* that host is selected as the target machine.
*/
char line[1024];
while( fgets(line, sizeof(line)-1, phostlist) != NULL ) {
elog(LOG, "HAWQ RM :: Load line %s", line);
/* Parse line. */
char *phostname = strtok(line, ",");
if ( phostname == NULL ) continue;
char *phostport = strtok(NULL, ",");
if ( phostport == NULL ) continue;
char *phostip = strtok(NULL, ",");
if ( phostip == NULL ) continue;
uint32_t port = 0;
if (sscanf(phostport, "%d", &port) != 1)
{
elog(LOG, "HAWQ RM :: Invalid port, skip machine %s", phostname);
continue;
}
/* If the host is new, add new host instance in resource pool */
int32_t segtid = SEGSTAT_ID_INVALID;
if ( getSegIDByHostName(phostname,strlen(phostname), &segtid) ==
FUNC_RETURN_OK )
{
elog(LOG, "HAWQ RM :: Skip machine %s", phostname);
continue;
}
elog(LOG, "HAWQ RM :: Find machine to add. %s", phostname);
/* Reserve fixed head part of the machine id instance. */
resetSelfMaintainBuffer(&seginfobuff);
prepareSelfMaintainBuffer(&seginfobuff, sizeof(SegInfoData),true);
SegInfo newhost = (SegInfo)(seginfobuff.Buffer);
newhost->master = 0;
newhost->standby = 0;
newhost->port = port;
newhost->alive = 1;
jumpforwardSelfMaintainBuffer(&seginfobuff, sizeof(SegInfoData));
/* Add address offset and attributes. Currently, only one address. */
newhost->HostAddrCount = 1;
newhost->AddressAttributeOffset = sizeof(SegInfoData);
/* Jump offset and reserved pad. */
newhost->AddressContentOffset = sizeof(SegInfoData) +
sizeof(uint32_t) * (((newhost->HostAddrCount + 1) >> 1) << 1);
uint16_t addrattr = HOST_ADDRESS_CONTENT_STRING;
uint16_t addroffset = newhost->AddressContentOffset;
appendSMBVar(&seginfobuff, addroffset);
appendSMBVar(&seginfobuff, addrattr);
appendSelfMaintainBufferTill64bitAligned(&seginfobuff);
/* Add hostip content. */
uint32_t length = strlen(phostip);
while( length > 0 &&
(phostip[length-1] == '\r' ||
phostip[length-1] == '\n' ||
phostip[length-1] == '\t' ||
phostip[length-1] == ' ') ) {
length--;
}
int addrsize = __SIZE_ALIGN64(offsetof(AddressStringData, Address) +
length + 1);
prepareSelfMaintainBuffer(&seginfobuff, addrsize, true);
AddressString straddr = (AddressString)(seginfobuff.Buffer +
seginfobuff.Cursor + 1);
straddr->Length = length;
memcpy(straddr->Address, phostip, length+1);
straddr->Address[length] = '\0';
jumpforwardSelfMaintainBuffer(&seginfobuff, addrsize);
newhost = (SegInfo)(seginfobuff.Buffer);
/* Append hostname string. */
newhost->HostNameLen = strlen(phostname);
newhost->HostNameOffset = seginfobuff.Cursor + 1;
appendSMBStr(&seginfobuff,phostname);
appendSelfMaintainBufferTill64bitAligned(&seginfobuff);
/* Update total machine id instance size. */
newhost = (SegInfo)(seginfobuff.Buffer);
newhost->Size = seginfobuff.Cursor + 1;
/* Build machine info instance. */
SegStat segstat = (SegStat)rm_palloc0(PCONTEXT,
offsetof(SegStatData, Info) +
seginfobuff.Cursor + 1);
segstat->ID = SEGSTAT_ID_INVALID;
segstat->FTSAvailable = RESOURCE_SEG_STATUS_AVAILABLE;
segstat->FTSTotalMemoryMB = DRMGlobalInstance->SegmentMemoryMB;
segstat->FTSTotalCore = DRMGlobalInstance->SegmentCore;
segstat->GRMTotalMemoryMB = 0;
segstat->GRMTotalCore = 0;
segstat->FailedTmpDirNum = 0;
memcpy((char *)segstat + offsetof(SegStatData, Info),
seginfobuff.Buffer,
seginfobuff.Cursor+1);
SelfMaintainBufferData segreport;
initializeSelfMaintainBuffer(&segreport,PCONTEXT);
generateSegStatReport(segstat, &segreport);
elog(LOG, "Resource manager builds available segment from file: %s",
segreport.Buffer);
destroySelfMaintainBuffer(&segreport);
bool capstatchanged = false;
res = addHAWQSegWithSegStat(segstat, &capstatchanged);
if ( res != FUNC_RETURN_OK )
{
elog(WARNING, "Resource manager failed to add machine from file.");
rm_pfree(PCONTEXT, segstat);
}
/* Fill HDFS and GRM host name cache */
int32_t id = SEGSTAT_ID_INVALID;
SimpString key;
setSimpleStringRefNoLen(&key, GET_SEGINFO_HOSTNAME(&(segstat->Info)));
getSegIDByHostName(GET_SEGINFO_HOSTNAME(&(segstat->Info)),
segstat->Info.HostNameLen,
&id);
setHASHTABLENode(&(PRESPOOL->HDFSHostNameIndexed),
&key,
TYPCONVERT(void *, id),
false);
setHASHTABLENode(&(PRESPOOL->GRMHostNameIndexed),
&key,
TYPCONVERT(void *, id),
false);
}
/* Refresh resource queue capacities. */
refreshResourceQueueCapacity(false);
refreshActualMinGRMContainerPerSeg();
/* Recalculate all memory/core ratio instances' limits. */
refreshMemoryCoreRatioLimits();
/* Refresh memory/core ratio level water mark. */
refreshMemoryCoreRatioWaterMark();
/* Free buffer if used. */
destroySelfMaintainBuffer(&seginfobuff);
fclose(phostlist);
return FUNC_RETURN_OK;
}
extern Datum dump_resource_manager_status(PG_FUNCTION_ARGS)
{
static char errorbuf[ERRORMESSAGE_SIZE];
int type = PG_GETARG_INT32(0);
char message[1024] = {0};
char dump_file[1024] = {0};
switch (type)
{
case 1:
strcpy(dump_file, "/tmp/resource_manager_conntrack_status");
sprintf(message, "Dump resource manager connection track status to %s", dump_file);
break;
case 2:
strcpy(dump_file, "/tmp/resource_manager_resqueue_status");
sprintf(message, "Dump resource manager resource queue status to %s", dump_file);
break;
case 3:
strcpy(dump_file, "/tmp/resource_manager_respool_status");
sprintf(message, "Dump resource manager resource pool status to %s", dump_file);
break;
default:
sprintf(message, "Dump resource manager status failed.\n"
"1 -> dump resource manager connection track status\n"
"2 -> dump resource manager resource queue status\n"
"3 -> dump resource manager resource pool status");
PG_RETURN_TEXT_P(cstring_to_text(message));
}
dumpResourceManagerStatus(type, dump_file, errorbuf, sizeof(errorbuf));
PG_RETURN_TEXT_P(cstring_to_text(message));
}
void processResourceBrokerTasks(void)
{
uint64_t curtime = 0;
int res = FUNC_RETURN_OK;
if ( !isCleanGRMResourceStatus() )
{
/*
* STEP 1. Check if should generate request to periodically check
* GRM cluster status.
*
* Generate request to the global resource manager ( YARN, etc.) to get
* cluster report. This is a asynchronous request, therefore, we don't
* need to handle the real cluster report right now. The possibility is
* reserved that if one fake global resource manager is implemented in
* HAWQ RM process and the cluster report can be returned directly in
* this function.
*/
curtime = gettime_microsec();
if ( (PRESPOOL->Segments.NodeCount > 0 ) &&
(curtime - PRESPOOL->LastUpdateTime >
rm_cluster_report_period * 1000000LL ||
requireInstantClusterReport() ) &&
(curtime - PRESPOOL->LastRequestTime > 5LL * 1000000LL) )
{
double maxcap = 0.0;
List *report = NULL;
PRESPOOL->LastRequestTime = curtime;
res = RB_getClusterReport(rm_grm_yarn_queue, &report, &maxcap);
Assert(report == NULL);
if ( res != FUNC_RETURN_OK )
{
elog(WARNING, "Resource manager fails to fresh cluster report "
"from global resource manager.");
goto exit;
}
}
/* STEP 2. Check if should periodically check container status. */
curtime = gettime_microsec();
if ( PRESPOOL->AddPendingContainerCount == 0 &&
PRESPOOL->RetPendingContainerCount == 0 &&
(curtime - PRESPOOL->LastCheckContainerTime >
rm_cluster_report_period * 1000000LL) &&
(curtime - PRESPOOL->LastRequestContainerTime > 5LL * 1000000LL) )
{
List *report = NULL;
PRESPOOL->LastRequestContainerTime = curtime;
res = RB_getContainerReport(&report);
Assert( report == NULL );
if ( res != FUNC_RETURN_OK )
{
elog(WARNING, "Resource manager fails to get container report "
"from global resource manager.");
goto exit;
}
}
/* STEP 3. Return kicked GRM containers. */
curtime = gettime_microsec();
if ( !PRESPOOL->pausePhase[QUOTA_PHASE_KICKED_TO_RETURN] )
{
res = RB_returnResource(&(PRESPOOL->KickedContainers));
if ( res != FUNC_RETURN_OK )
{
elog(WARNING, "Resource manager failed to return kicked container to "
"global resource manager.");
goto exit;
}
}
else
{
elog(LOG, "Paused returning GRM containers kicked to GRM.");
}
/*
* STEP 4. Handle resource broker input as new allocated resource or
* cluster report, container report etc. The allocated resource
* from resource broker will be added to the resource queue
* manager and resource pool.
*/
res = RB_handleNotifications();
if ( res != FUNC_RETURN_OK )
{
elog(WARNING, "Resource manager fails to handle response from global "
"resource broker");
goto exit;
}
}
else
{
RB_clearResource(&(PRESPOOL->KickedContainers));
}
exit:
RB_handleError(res);
}
void generateResourceRequestToResourceBroker(void)
{
uint64_t curtime = 0;
int res = FUNC_RETURN_OK;
if ( !isCleanGRMResourceStatus() )
{
curtime = gettime_microsec();
if ( PRESPOOL->Segments.NodeCount > 0 && PQUEMGR->RatioCount > 0 )
{
refreshMemoryCoreRatioLevelUsage(curtime);
if ( curtime - PRESPOOL->LastResAcqTime > 1000000LL)
{
res = generateAllocRequestToBroker();
if ( res != FUNC_RETURN_OK )
{
elog(WARNING, "Resource manager fails to allocate container "
"from global resource manager.");
goto exit;
}
}
}
}
exit:
RB_handleError(res);
}
void cleanupAllGRMContainers(void)
{
dropAllResPoolGRMContainersToToBeKicked();
dropAllToAcceptGRMContainersToKicked();
}
bool cleanedAllGRMContainers(void)
{
/* Condition 1. All segments have no resource allocated. */
if ( !allSegmentHasNoGRMContainersAllocated() )
{
return false;
}
if ( PRESPOOL->AddPendingContainerCount > 0 ||
PRESPOOL->RetPendingContainerCount > 0 )
{
elog(RMLOG, "Pending GRM container count inc %d, dec %d.",
PRESPOOL->AddPendingContainerCount,
PRESPOOL->RetPendingContainerCount);
}
/* Condition 2. No on-the-fly GRM containers for increasing and decreasing.*/
return PRESPOOL->AddPendingContainerCount == 0 &&
PRESPOOL->RetPendingContainerCount == 0;
}