资源控制在大数据和云计算平台中的应用

[root@docker ~]# mount | grep cgroup
tmpfs on /sys/fs/cgroup type tmpfs (ro,nosuid,nodev,noexec,mode=755)
cgroup on /sys/fs/cgroup/systemd type cgroup (rw,nosuid,nodev,noexec,relatime,xattr,release_agent=/usr/lib/systemd/systemd-cgroups-agent,name=systemd)
cgroup on /sys/fs/cgroup/cpu,cpuacct type cgroup (rw,nosuid,nodev,noexec,relatime,cpuacct,cpu)
cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blkio)
cgroup on /sys/fs/cgroup/hugetlb type cgroup (rw,nosuid,nodev,noexec,relatime,hugetlb)
cgroup on /sys/fs/cgroup/perf_event type cgroup (rw,nosuid,nodev,noexec,relatime,perf_event)
cgroup on /sys/fs/cgroup/pids type cgroup (rw,nosuid,nodev,noexec,relatime,pids)
cgroup on /sys/fs/cgroup/cpuset type cgroup (rw,nosuid,nodev,noexec,relatime,cpuset)
cgroup on /sys/fs/cgroup/net_cls,net_prio type cgroup (rw,nosuid,nodev,noexec,relatime,net_prio,net_cls)
cgroup on /sys/fs/cgroup/memory type cgroup (rw,nosuid,nodev,noexec,relatime,memory)
cgroup on /sys/fs/cgroup/freezer type cgroup (rw,nosuid,nodev,noexec,relatime,freezer)
cgroup on /sys/fs/cgroup/devices type cgroup (rw,nosuid,nodev,noexec,relatime,devices)

[root@docker sym]# mkdir -p /sys/fs/cgroup/memory/ibm/symphony
[root@docker sym]# ls /sys/fs/cgroup/memory/ibm/symphony
cgroup.clone_children  memory.kmem.failcnt             memory.kmem.tcp.limit_in_bytes      memory.max_usage_in_bytes        memory.move_charge_at_immigrate  memory.stat            tasks
cgroup.event_control   memory.kmem.limit_in_bytes      memory.kmem.tcp.max_usage_in_bytes  memory.memsw.failcnt             memory.numa_stat                 memory.swappiness
cgroup.procs           memory.kmem.max_usage_in_bytes  memory.kmem.tcp.usage_in_bytes      memory.memsw.limit_in_bytes      memory.oom_control               memory.usage_in_bytes
memory.failcnt         memory.kmem.slabinfo            memory.kmem.usage_in_bytes          memory.memsw.max_usage_in_bytes  memory.pressure_level            memory.use_hierarchy
memory.force_empty     memory.kmem.tcp.failcnt         memory.limit_in_bytes               memory.memsw.usage_in_bytes      memory.soft_limit_in_bytes       notify_on_release

[root@docker sym]# echo 23790 >> /sys/fs/cgroup/memory/ibm/symphony/tasks

[root@docker sym]# echo 1024M > /sys/fs/cgroup/memory/ibm/symphony/memory.limit_in_bytes 
[root@docker sym]# cat /sys/fs/cgroup/memory/ibm/symphony/memory.limit_in_bytes
1073741824

[root@docker sym]# cat /sys/fs/cgroup/memory/memory.limit_in_bytes 
9223372036854771712

#include "stdafx.h"
#include <stdio.h>
#include <iostream>
#include <ace/ACE.h>  
#include <ace/Init_ACE.h>  
#include <ace/Process.h> 
#include <windows.h>
using namespace std;
HANDLE getJobObj(const HANDLE& processHandle, const HANDLE& threadHandle)
{
    auto jobName = _T("MyJob");
    HANDLE jobHandle = NULL;
    bool success = false;
    //////////////////////////////////////////////////////////////////////////
    // If process is already in Job, we use OpenJobObject instead of CreateJobObject
    //////////////////////////////////////////////////////////////////////////
    BOOL injob = FALSE;
    IsProcessInJob(processHandle, NULL, &injob);
    if (injob)
    {
        // 如果已经属于一个内核对象，直接打开
        jobHandle = OpenJobObject(JOB_OBJECT_ALL_ACCESS, FALSE, jobName);
    }
    else
    {
        //创建一个新的作业内核对象
        jobHandle = CreateJobObject(NULL, jobName);
    }
     //
    //////////////////////////////////////////////////////////////////////////
    // 关联完成端口
    //////////////////////////////////////////////////////////////////////////
    auto hIOCP = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
    JOBOBJECT_ASSOCIATE_COMPLETION_PORT joacp;
    joacp.CompletionKey = (PVOID)1; // Any value to uniquely identify this job
    joacp.CompletionPort = hIOCP; // Handle of completion port that
    // receives notifications
    SetInformationJobObject(jobHandle,
        JobObjectAssociateCompletionPortInformation,
        &joacp,
        sizeof(joacp));
    //////////////////////////////////////////////////////////////////////////
    //为作业添加一些基本限制
    //////////////////////////////////////////////////////////////////////////
    JOBOBJECT_EXTENDED_LIMIT_INFORMATION extendLimit = { 0 };
    extendLimit.BasicLimitInformation.LimitFlags = 0;
    // 作业物理内存 500M
    extendLimit.BasicLimitInformation.LimitFlags |= JOB_OBJECT_LIMIT_WORKINGSET;
    extendLimit.BasicLimitInformation.MaximumWorkingSetSize = 1024 * 1024 * 500;
    extendLimit.BasicLimitInformation.MinimumWorkingSetSize = 64;
    std::cout << "Set Physical MEM = " << extendLimit.BasicLimitInformation.MaximumWorkingSetSize / 1024 / 1024 << " M" << std::endl;
    // 作业虚拟内存 700M
    extendLimit.BasicLimitInformation.LimitFlags |= JOB_OBJECT_LIMIT_JOB_MEMORY;
    extendLimit.JobMemoryLimit = 1024 * 1024 * 700;
    std::cout << "Set Virtual MEM = " << extendLimit.JobMemoryLimit/1024/1024 << " M"<< std::endl;
    //设定作业限制
    success = SetInformationJobObject(jobHandle, JobObjectExtendedLimitInformation, &extendLimit, sizeof(extendLimit));
    if (!success)
    {
        std::cout << "Set JobObjectExtendedLimitInformation Failed :" << GetLastError() << std::endl;
    }
    // CPU 使用率
    JOBOBJECT_CPU_RATE_CONTROL_INFORMATION cpuRate = { 0 };
    cpuRate.ControlFlags = 0;
    cpuRate.ControlFlags |= JOB_OBJECT_CPU_RATE_CONTROL_ENABLE;
    cpuRate.ControlFlags |= JOB_OBJECT_CPU_RATE_CONTROL_HARD_CAP;
    cpuRate.CpuRate = 100 * 20; // 20%
    success = SetInformationJobObject(jobHandle, JobObjectCpuRateControlInformation, &cpuRate, sizeof(cpuRate));
    if (!success)
    {
        std::cout << "Set JobObjectCpuRateControlInformation Failed :" << GetLastError() << std::endl;
    }
    //将进程添加到作业
    success = AssignProcessToJobObject(jobHandle, processHandle);
    if (!success)
    {
        std::cout << "AssignProcessToJobObject Failed :" << GetLastError() << std::endl;
    }
    //唤醒进程（的主线程）
    ResumeThread(threadHandle);
    //////////////////////////////////////////////////////////////////////////
    // Check IO Completion
    //////////////////////////////////////////////////////////////////////////
    DWORD dwReasonID = 0; //事件ID，参数lpNumberOfBytes
    DWORD dwProcessID = 0;
    OVERLAPPED* lpOverlapped = NULL;
    DWORD dwMilliseconds = 1000 * 3;// INFINITE
    if (!GetQueuedCompletionStatus(hIOCP, &dwReasonID,
        (PULONG_PTR)&jobHandle, &lpOverlapped, dwMilliseconds))
    {
        printf(("IOCPThread:GetQueueCompletionStatus调用失败，错误代码：0x%08xn"),
            GetLastError());
    }
    else
    {
        //checkJobEvent(dwReasonID);
    }
    Sleep(1000 * 10);
    //////////////////////////////////////////////////////////////////////////
    // 关闭IO完成端口
    //////////////////////////////////////////////////////////////////////////
    CloseHandle(hIOCP);
    return jobHandle;
}
class MyProcess : public ACE_Process
{
public:
    MyProcess::MyProcess() : m_jobHandle(NULL){}
    virtual MyProcess::~MyProcess() { if (m_jobHandle != NULL) CloseHandle(m_jobHandle); }
    // return process thread handle
    HANDLE MyProcess::getThreadHandle() { return process_info_.hThread; }
    // Save windows job object handle for close when process gone.
    void MyProcess::jobHandle(HANDLE& handle) { m_jobHandle = handle; }
    // For some other reason, we need detach job handle
    virtual void MyProcess::detach() { m_jobHandle = NULL; }
    // Close job object when terminating.
    virtual void MyProcess::terminate()
    {
        ACE_Process::terminate();
        if (m_jobHandle != NULL)
        {
            CloseHandle(m_jobHandle);
            m_jobHandle = NULL;
        }
    }
private:
    HANDLE m_jobHandle;
};
int _tmain(int argc, _TCHAR* argv[])
{
    ACE::init();
    MyProcess p;
    ACE_Process_Options option;
    option.command_line(_T("C:\Windows\System32\notepad.exe"));
    option.creation_flags(CREATE_SUSPENDED | CREATE_BREAKAWAY_FROM_JOB);
    auto pid = p.spawn(option);
    auto handle = getJobObj(p.gethandle(), p.getThreadHandle());
    p.jobHandle(handle);
    std::cout << "PID:" << pid << endl;
    pid = p.wait();
    std::cout << "Wait:" << pid << endl;
    getchar();
    p.terminate();
    getchar();
    ACE::fini();
    return 0;
}

[root@rhel711 ~]# docker run -ti --rm ubuntu /bin/sh
[root@rhel711 ~]# docker ps -a
CONTAINER ID        IMAGE               COMMAND             CREATED              STATUS              PORTS               NAMES
e9102a6edb29        ubuntu              "/bin/sh"           About a minute ago   Up About a minute                       trusting_ritchie
[root@rhel711 ~]# ls /sys/fs/cgroup/memory/system.slice/docker-e9102a6edb29491ee92ed0441238d41ca7c3d5d5841757679d17025ffa4b9f39.scope/
cgroup.clone_children       memory.kmem.max_usage_in_bytes      memory.limit_in_bytes            memory.numa_stat            memory.use_hierarchy
cgroup.event_control        memory.kmem.slabinfo                memory.max_usage_in_bytes        memory.oom_control          notify_on_release
cgroup.procs                memory.kmem.tcp.failcnt             memory.memsw.failcnt             memory.pressure_level       tasks
memory.failcnt              memory.kmem.tcp.limit_in_bytes      memory.memsw.limit_in_bytes      memory.soft_limit_in_bytes
memory.force_empty          memory.kmem.tcp.max_usage_in_bytes  memory.memsw.max_usage_in_bytes  memory.stat
memory.kmem.failcnt         memory.kmem.tcp.usage_in_bytes      memory.memsw.usage_in_bytes      memory.swappiness
memory.kmem.limit_in_bytes  memory.kmem.usage_in_bytes          memory.move_charge_at_immigrate  memory.usage_in_bytes

[root@rhel711 ~]# cat /sys/fs/cgroup/memory/system.slice/docker-e9102a6edb29491ee92ed0441238d41ca7c3d5d5841757679d17025ffa4b9f39.scope/tasks 
27230
[root@rhel711 ~]# ps -ef | grep 27230
root     27230  1182  0 02:39 pts/3    00:00:00 /bin/sh

[root@docker sym]# docker exec -ti tensorflow cat /proc/mounts | grep memory
cgroup /sys/fs/cgroup/memory cgroup rw,nosuid,nodev,noexec,relatime,memory 0 0

    spec:
      containers:
        - name: sym-compute
          image: sequenceiq/spark
          command: ["/bin/sh", "-c", "start.sh"]
          resources:
            requests:
              memory: 4096M
            limits:
              memory: 4096M