Greenplum 资源管理——Resource Group使用和实现分析

CREATE RESOURCE GROUP rgroup1 WITH (
  CONCURRENCY=10,
  CPU_RATE_LIMIT=20,
  MEMORY_LIMIT=25,
  MEMORY_SHARED_QUOTA=20,
  MEMORY_SPILL_RATIO=20
);

CREATE RESOURCE GROUP rgroup1 WITH (CPU_RATE_LIMIT=20, MEMORY_LIMIT=25);

ALTER ROLE bill RESOURCE GROUP rg_light;
CREATE ROLE mary RESOURCE GROUP exec;

ALTER ROLE mary RESOURCE GROUP NONE;

ALTER RESOURCE GROUP rg_role_light SET CONCURRENCY 7;
ALTER RESOURCE GROUP exec SET MEMORY_LIMIT 25;
ALTER RESOURCE GROUP rgroup1 SET CPUSET '2,4';

DROP RESOURCE GROUP exec;

SELECT rolname, g.rsgname, procpid, waiting, current_query, datname
     FROM pg_roles, gp_toolkit.gp_resgroup_status g, pg_stat_activity
     WHERE pg_roles.rolresgroup=g.groupid
        AND pg_stat_activity.usename=pg_roles.rolname;

 rolname | rsgname  | procpid | waiting |     current_query     | datname
---------+----------+---------+---------+-----------------------+---------
  sammy  | rg_light |  31861  |    f    | <IDLE> in transaction | testdb
  billy  | rg_light |  31905  |    t    | SELECT * FROM topten; | testdb

SELECT pg_cancel_backend(31905);

memory_used = memory_quota_used + memory_shared_used
available = memory_quota_granted + memory_shared_granted - memory_used
memory_quota_granted = memory_quota_used + memory_quota_available
memory_shared_granted = memory_shared_used + memory_shared_available

SELECT * FROM gp_toolkit.gp_resgroup_config;

SELECT * FROM gp_toolkit.gp_resgroup_status;

SELECT * FROM gp_toolkit.gp_resgroup_status_per_host;

SELECT * FROM gp_toolkit.gp_resgroup_status_per_segment;

SELECT rolname, rsgname FROM pg_roles, pg_resgroup
     WHERE pg_roles.rolresgroup=pg_resgroup.oid;

SELECT current_query, waiting, rsgname, rsgqueueduration
     FROM pg_stat_activity;

src/backend/utils/resgroup
src/backend/utils/ressource_manager
src/backend/commands/resgroupcmds.c

static ResGroupSlotData *
groupGetSlot(ResGroupData *group)
{
        ResGroupSlotData	*slot;
        ResGroupCaps		*caps;
        int32				slotMemQuota;
        Assert(LWLockHeldExclusiveByMe(ResGroupLock));
        Assert(Gp_role == GP_ROLE_DISPATCH);
        Assert(groupIsNotDropped(group));
        caps = &group->caps;
        /* First check if the concurrency limit is reached */
        if (group->nRunning >= caps->concurrency)
                return NULL;
        slotMemQuota = groupReserveMemQuota(group);
        if (slotMemQuota < 0)
                return NULL;
        /* Now actually get a free slot */
        slot = slotpoolAllocSlot();
        Assert(!slotIsInUse(slot));
        initSlot(slot, group, slotMemQuota);
        group->nRunning++;
        return slot;
}

void
ResGroupOps_AssignGroup(Oid group, ResGroupCaps *caps, int pid)
{
        bool oldViaCpuset = oldCaps.cpuRateLimit == CPU_RATE_LIMIT_DISABLED;
        bool curViaCpuset = caps ? caps->cpuRateLimit == CPU_RATE_LIMIT_DISABLED : false;
        /* needn't write to file if the pid has already been written in.
         * Unless it has not been writtien or the group has changed or
         * cpu control mechanism has changed */
        if (IsUnderPostmaster &&
                group == currentGroupIdInCGroup &&
                caps != NULL &&
                oldViaCpuset == curViaCpuset)
                return;
        writeInt64(group, BASETYPE_GPDB, RESGROUP_COMP_TYPE_CPU,
                           "cgroup.procs", pid);
        writeInt64(group, BASETYPE_GPDB, RESGROUP_COMP_TYPE_CPUACCT,
                           "cgroup.procs", pid);
        if (gp_resource_group_enable_cgroup_cpuset)
        {
                if (caps == NULL || !curViaCpuset)
                {
                        /* add pid to default group */
                        writeInt64(DEFAULT_CPUSET_GROUP_ID, BASETYPE_GPDB,
                                           RESGROUP_COMP_TYPE_CPUSET, "cgroup.procs", pid);
                }
                else
                {
                        writeInt64(group, BASETYPE_GPDB,
                                           RESGROUP_COMP_TYPE_CPUSET, "cgroup.procs", pid);
                }
        }
        /*
         * Do not assign the process to cgroup/memory for now.
         */
        currentGroupIdInCGroup = group;
        if (caps != NULL)
        {
                oldCaps.cpuRateLimit = caps->cpuRateLimit;
                StrNCpy(oldCaps.cpuset, caps->cpuset, sizeof(oldCaps.cpuset));
        }
}

static int32
groupIncMemUsage(ResGroupData *group, ResGroupSlotData *slot, int32 chunks)
{
        int32			slotMemUsage;	/* the memory current slot has been used */
        int32			sharedMemUsage;	/* the total shared memory usage,
                                                                                sum of group share and global share */
        int32			globalOveruse = 0;	/* the total over used chunks of global share*/
        /* Add the chunks to memUsage in slot */
        slotMemUsage = pg_atomic_add_fetch_u32((pg_atomic_uint32 *) &slot->memUsage,
                                                                                   chunks);
        /* Check whether shared memory should be added */
        sharedMemUsage = slotMemUsage - slot->memQuota;
        if (sharedMemUsage > 0)
        {
                /* Decide how many chunks should be counted as shared memory */
                int32 deltaSharedMemUsage = Min(sharedMemUsage, chunks);
                /* Add these chunks to memSharedUsage in group,
                 * and record the old value*/
                int32 oldSharedUsage = pg_atomic_fetch_add_u32((pg_atomic_uint32 *)
                                                                                                           &group->memSharedUsage,
                                                                                                           deltaSharedMemUsage);
                /* the free space of group share */
                int32 oldSharedFree = Max(0, group->memSharedGranted - oldSharedUsage);
                /* Calculate the global over used chunks */
                int32 deltaGlobalSharedMemUsage = Max(0, deltaSharedMemUsage - oldSharedFree);
                /* freeChunks -= deltaGlobalSharedMemUsage and get the new value */
                int32 newFreeChunks = pg_atomic_sub_fetch_u32((pg_atomic_uint32 *)
                                                                                                          &pResGroupControl->freeChunks,
                                                                                                          deltaGlobalSharedMemUsage);
                /* calculate the total over used chunks of global share */
                globalOveruse = Max(0, 0 - newFreeChunks);
        }
        /* Add the chunks to memUsage in group */
        pg_atomic_add_fetch_u32((pg_atomic_uint32 *) &group->memUsage,
                                                        chunks);
        return globalOveruse;
}

CREATE VIEW gp_toolkit.gp_resgroup_status AS
    SELECT r.rsgname, s.*
    FROM pg_resgroup_get_status(null) AS s,
         pg_resgroup AS r
    WHERE s.groupid = r.oid;

static void
getResUsage(ResGroupStatCtx *ctx, Oid inGroupId)
{
        int64 *usages;
        TimestampTz *timestamps;
        int i, j;
        usages = palloc(sizeof(*usages) * ctx->nGroups);
        timestamps = palloc(sizeof(*timestamps) * ctx->nGroups);
        for (j = 0; j < ctx->nGroups; j++)
        {
                ResGroupStat *row = &ctx->groups[j];
                Oid groupId = DatumGetObjectId(row->groupId);
                usages[j] = ResGroupOps_GetCpuUsage(groupId);
                timestamps[j] = GetCurrentTimestamp();
        }
        if (Gp_role == GP_ROLE_DISPATCH)
        {
                CdbPgResults cdb_pgresults = {NULL, 0};
                StringInfoData buffer;
                initStringInfo(&buffer);
                appendStringInfo(&buffer,
                                                 "SELECT groupid, cpu_usage, memory_usage "
                                                 "FROM pg_resgroup_get_status(%d)",
                                                 inGroupId);
                CdbDispatchCommand(buffer.data, DF_WITH_SNAPSHOT, &cdb_pgresults);
                if (cdb_pgresults.numResults == 0)
                        elog(ERROR, "pg_resgroup_get_status() didn't get back any resource statistic from the segDBs");
                for (i = 0; i < cdb_pgresults.numResults; i++)
                {
                        struct pg_result *pg_result = cdb_pgresults.pg_results[i];
                        /*
                         * Any error here should have propagated into errbuf, so we shouldn't
                         * ever see anything other that tuples_ok here.  But, check to be
                         * sure.
                         */
                        if (PQresultStatus(pg_result) != PGRES_TUPLES_OK)
                        {
                                cdbdisp_clearCdbPgResults(&cdb_pgresults);
                                elog(ERROR, "pg_resgroup_get_status(): resultStatus not tuples_Ok");
                        }
                        Assert(PQntuples(pg_result) == ctx->nGroups);
                        for (j = 0; j < ctx->nGroups; j++)
                        {
                                const char *result;
                                ResGroupStat *row = &ctx->groups[j];
                                Oid groupId = pg_atoi(PQgetvalue(pg_result, j, 0),
                                                                          sizeof(Oid), 0);
                                Assert(groupId == row->groupId);
                                if (row->memUsage->len == 0)
                                {
                                        Datum d = ResGroupGetStat(groupId, RES_GROUP_STAT_MEM_USAGE);
                                        row->groupId = groupId;
                                        appendStringInfo(row->memUsage, "{\"%d\":%s",
                                                                         GpIdentity.segindex, DatumGetCString(d));
                                        appendStringInfo(row->cpuUsage, "{");
                                        calcCpuUsage(row->cpuUsage, usages[j], timestamps[j],
                                                                 ResGroupOps_GetCpuUsage(groupId),
                                                                 GetCurrentTimestamp());
                                }
                                result = PQgetvalue(pg_result, j, 1);
                                appendStringInfo(row->cpuUsage, ", %s", result);
                                result = PQgetvalue(pg_result, j, 2);
                                appendStringInfo(row->memUsage, ", %s", result);
                                if (i == cdb_pgresults.numResults - 1)
                                {
                                        appendStringInfoChar(row->cpuUsage, '}');
                                        appendStringInfoChar(row->memUsage, '}');
                                }
                        }
                }
                cdbdisp_clearCdbPgResults(&cdb_pgresults);
        }
        else
        {
                pg_usleep(300000);
                for (j = 0; j < ctx->nGroups; j++)
                {
                        ResGroupStat *row = &ctx->groups[j];
                        Oid groupId = DatumGetObjectId(row->groupId);
                        Datum d = ResGroupGetStat(groupId, RES_GROUP_STAT_MEM_USAGE);
                        appendStringInfo(row->memUsage, "\"%d\":%s",
                                                         GpIdentity.segindex, DatumGetCString(d));
                        calcCpuUsage(row->cpuUsage, usages[j], timestamps[j],
                                                 ResGroupOps_GetCpuUsage(groupId),
                                                 GetCurrentTimestamp());
                }
        }
}

static void
calcCpuUsage(StringInfoData *str,
                         int64 usageBegin, TimestampTz timestampBegin,
                         int64 usageEnd, TimestampTz timestampEnd)
{
        int64 duration;
        long secs;
        int usecs;
        int64 usage;
        usage = usageEnd - usageBegin;
        TimestampDifference(timestampBegin, timestampEnd, &secs, &usecs);
        duration = secs * 1000000 + usecs;
        appendStringInfo(str, "\"%d\":%.2f",
                                         GpIdentity.segindex,
                                         ResGroupOps_ConvertCpuUsageToPercent(usage, duration));
}

group gpdb {
     perm {
         task {
             uid = gpadmin;
             gid = gpadmin;
         }
         admin {
             uid = gpadmin;
             gid = gpadmin;
         }
     }
     cpu {
     }
     cpuacct {
     }
     cpuset {
     }
     memory {
     }
}