OpenHarmony 资源调度之内存管理源码分析

作者：张守忠

1 内存管理简介

内存管理部件位于全局资源调度管控子系统中，基于应用的生命周期状态，更新进程回收优先级列表，通过内存回收、查杀等手段管理系统内存，保障内存供给。

1.1 内存管理框架

内存管理部件主要基于事件触发回收优先级更新，并基于回收优先级管理内存回收和查杀，其框架如下所示（下图虚线右侧为本部件），主要分为如下几个模块：

图 1 框架图

1、事件管理模块：统筹管理本部件所要对外响应的事件。主要功能为调用其他子系统提供的接口注册事件监听，并将事件通知到回收优先级管理模块、回收策略模块和查杀策略模块。

2、回收优先级管理模块：基于事件管理模块的应用和用户的事件通知，给出进程的回收和查杀的先后顺序列表，并向回收策略和查杀策略提供查询此列表的接口。

3、回收策略模块：根据回收优先级列表，调整回收内存水线、文件页/匿名页回收比例、压缩/换出比例等回收参数，以及协调不同回收机制协同工作，保障内存中低负载下的供给性能。

4、查杀策略模块：作为回收的逻辑末端，根据回收优先级列表，保障内存重负载下的内存供给。

5、Kernel接口管控模块：负责将回收参数、查杀等管控命令下发到Kernel。

6、内存特性配置：读取回收策略模块、查杀策略模块等需要的配置文件。

1.2 ESWAP简介

ESWAP（Enhanced SWAP）是 OpenHarmony 针对传统linux内存优化问题提供的一套完善的内存解决方案，结合内存压缩和内存交换技术，定制了一套合理高效的调度管理策略，使压缩和交换两者的工作能够高效且平衡。ESWAP 基于关联性的数据聚合技术及上层指导策略，将内存划分为不同的分组进行管理，通过回收优先级来区分不同分组下内存的活跃程度，优先压缩、换出较不活跃的内存数据，以提升数据交换性能，减少冲击。ESWAP 解决方案在全局资源调度子系统中增加了一个系统资源调度模块，通过向账户子系统订阅本地账户的变化来感知当前的账户状态和内存状态，然后根据账户状态给各个账户设置不同的回收优先级、设置目标可用内存量、设置压缩和换出的比例等参数，并将这些参数下发给 ZSWAPD。ZSWAPD 会依据回收优先级判断回收的先后顺序；依据目标可用内存量和当前可用内存量的差值决定回收的量；依据压缩和换出的比例来决定压缩和换出的量，从而实现在达成内存扩展效果前提下的性能和功耗平衡。 ESWAP方案涉及的关键技术：

定制的ZRAM和交换分区，ESWAP 结合内存压缩和内存交换技术，提供了自定义新增存储分区作为内存交换分区的能力，并在内核中创建了一个常驻进程 ZSWAPD，用于将 ZRAM 压缩后的匿名页加密换出到 ESWAP 存储分区中，从而能完全地空出一块可用内存，以此来达到维持 Memavailable 水线的目标。同时，ESWAP 模块还可以记录每个匿名页的冷热特征信息，并将这些数据通过关联性、冷热顺序进行相应的存放，使 ESWAP 交换区中连续存放的匿名页具有时间和空间局部性。因此在匿名页换入时，可以将交换区中的相邻匿名页一并读入 ZRAM，以此来保证数据的存取速度，提升 IO 性能。
动态的内存回收机制，提供了一种额外的内存回收机制 ZSWAPD，并创建了“buffer”来作为衡量当前系统内存能力的指标。buffer 指的是当前系统能提供的最大可用内存。ZSWAPD 会根据 buffer 量以及上文所述的各种策略，来对匿名页进行压缩换出以回收。
灵活的内存回收策略，基于 Memcg 分组进行了回收策略的增强，使用回收优先级来指导 ZSWAPD 回收的先后顺序。上层可以根据需要，通过灵活地配置交换策略，控制RAM、ZRAM、ESWAP 三个模块中存储的比例，避免频繁换入换出带来的负面影响。

1.3 PSI简介

PSI (Pressure Stall Information) 是 Facebook 在 2018 年开源的一套解决重要计算集群管理问题的 Linux 内核组件和相关工具中重要的资源度量工具，它提供了一种实时检测系统资源竞争程度的方法，以竞争等待时间的方式呈现，简单而准确地供用户以及资源调度者进行决策。PSI 软件结构图如下所示：

图 2 PSI软件结构图对上，PSI 模块通过文件系统节点向用户空间开放两种形态的接口。一种是系统级别的接口，即输出整个系统级别的资源压力信息。另外一种是结合 control group，进行更精细化的分组。对下，PSI 模块通过在内存管理模块以及调度器模块中插桩，我们可以跟踪每一个任务由于 memory、io 以及 CPU 资源而进入等待状态的信息。例如系统中处于 iowait 状态的 task 数目、由于等待 memory 资源而处于阻塞状态的任务数目。基于 task 维度的信息，PSI 模块会将其汇聚成 PSI group 上的 per cpu 维度的时间信息。例如该cpu上部分任务由于等待 IO 操作而阻塞的时间长度（CPU 并没有浪费，还有其他任务在执行）。PSI group 还会设定一个固定的周期去计算该采样周期内核的当前 psi 值（基于该 group 的 per cpu 时间统计信息）。为了避免 PSI 值的抖动，实际上上层应用通过系统调用获取某个 PSI group 的压力值的时候会上报近期一段时间值的滑动平均值。

1.3.1 PSI用户接口定义

每类资源的压力信息都通过 proc 文件系统的独立文件来提供，路径为 /proc/pressure/ – cpu, memory, and io.其中 CPU 压力信息格式如下：some avg10=2.98 avg60=2.81 avg300=1.41 total=268109926memory 和 io 格式如下：some avg10=0.30 avg60=0.12 avg300=0.02 total=4170757full avg10=0.12 avg60=0.05 avg300=0.01 total=1856503avg10、avg60、avg300 分别代表 10s、60s、300s 的时间周期内的阻塞时间百分比。total 是总累计时间，以毫秒为单位some 这一行，代表至少有一个任务在某个资源上阻塞的时间占比，full 这一行，代表所有的非idle任务同时被阻塞的时间占比，这期间 cpu 被完全浪费，会带来严重的性能问题。

2 关键实现

2.1 内存管理服务初始化

内存管理服务运行在进程memmgrservice中，在进程拉起时调用内存管理服务初始化程序，其初始化工作主要包括：配置文件解析、回收优先级初始化、更新账户优先级、回收策略初始化和各类监听事件注册，时序图如下：

2.2 回收优先级更新

回收优先级的更新涉及多个事件的触发，Account状态变化、应用状态变化、屏幕状态变化、电源状态等通用事件。

2.3 PSI上报事件处理

图 3 初始化时序图

系统能力注册在文件res_sched_service_ability.cpp中调用宏"REGISTER_SYSTEM_ABILITY_BY_ID"注册系统能力ResSchedServiceAbility

namespace OHOS {
namespace ResourceSchedule {
REGISTER_SYSTEM_ABILITY_BY_ID(ResSchedServiceAbility, RES_SCHED_SYS_ABILITY_ID, true);

系统能力初始化系统能力初始化分为资源调度管理初始化、系统服务发布、cpu分组调度初始化、注册监听的SA四个子过程。

void ResSchedServiceAbility::OnStart()
{
    ResSchedMgr::GetInstance().Init();
    if (!service_) {
        try {
            service_ = new ResSchedService();
        } catch(const std::bad_alloc &e) {
            RESSCHED_LOGE("ResSchedServiceAbility:: new ResSchedService failed.");
        }
    }
    if (!Publish(service_)) {
        RESSCHED_LOGE("ResSchedServiceAbility:: Register service failed.");
    }
    CgroupSchedInit();
    AddSystemAbilityListener(APP_MGR_SERVICE_ID);
    AddSystemAbilityListener(WINDOW_MANAGER_SERVICE_ID);
    AddSystemAbilityListener(BACKGROUND_TASK_MANAGER_SERVICE_ID);
    RESSCHED_LOGI("ResSchedServiceAbility ::OnStart.");
}

资源调度管理初始化过程主要包括frame_aware和socperf两个的插件加载及函数指针获取，eventHandler对象创建。插件so库的路径配置在xml文件"res_sched_plugin_switch.xml"中。

cpu分组调度初始化主要分为supervisor、cgroupHandler和cgroupAdjuster三个子模块的初始化。

void SchedController::Init()
{
    ChronoScope cs("Init SchedController.");
    // Init supervisor which contains cached data for ccgroup controller.
    InitSupervisor();
    // Init cgroup handler thread
    InitCgroupHandler();
    // Init cgroup adjuster thread
    InitCgroupAdjuster();
}

注册监听的SA调用SA框架提供的接口完成监听SA的注册, 目前监听了三个SA，分别是应用管理服务(id: APP_MGR_SERVICE_ID)、窗口管理服务(id: WINDOW_MANAGER_SERVICE_ID)和后台任务管理服务(id: BACKGROUND_TASK_MANAGER_SERVICE_ID).

bool SystemAbility::AddSystemAbilityListener(int32_t systemAbilityId)
{
    HILOGD(TAG, "SA:%{public}d, listenerSA:%{public}d", systemAbilityId, saId_);
    return LocalAbilityManager::GetInstance().AddSystemAbilityListener(systemAbilityId, saId_);
}

bool LocalAbilityManager::AddSystemAbilityListener(int32_t systemAbilityId, int32_t listenerSaId)
{
    ...
   
    auto samgrProxy = SystemAbilityManagerClient::GetInstance().GetSystemAbilityManager();
    if (samgrProxy == nullptr) {
        HILOGE(TAG, "failed to get samgrProxy");
        return false;
    }

    {
        ...
        
        auto& listenerSaIdList = listenerMap_[systemAbilityId];
        auto iter = std::find_if(listenerSaIdList.begin(), listenerSaIdList.end(), [listenerSaId](int32_t SaId) {
            return SaId == listenerSaId;
        });
        if (iter == listenerSaIdList.end()) {
            listenerSaIdList.emplace_back(listenerSaId);
        }
        ...
    }

    int32_t ret = samgrProxy->SubscribeSystemAbility(systemAbilityId, GetSystemAbilityStatusChange());
    if (ret) {
        HILOGE(TAG, "failed to subscribe sa:%{public}d, process name:%{public}s", systemAbilityId,
            Str16ToStr8(procName_).c_str());
        return false;
    }
    return true;
}

2.3 绘帧信息收集流程

图 4 绘帧信息收集资源调度时序图
绘帧信息收集主要根据场景分为输入子系统手势变化等的感知调度，ace子系统的窗口变化等的感知调度，图形子系统的渲染、动画感知调度。图4描述的是BeginListFling、BeginFlushBuild、Render三个绘帧信息资源调度时序图，其他的流程类似，不一一列举。绘帧信息收集资源调度的最终是通过内核接口ioctrl写关联线程组"/dev/sched_rtg_ctrl", 包括线程加入、移除关联线程组，设置关联线程组的属性等等。

2.4 帧感知调度流程

被监听的SA的相关属性或状态变化，调用订阅的对象接口进行响应。

该对象通过eventHandler机制在event runner线程中调整进程组, 通过进程间通讯IPC通知资源调度服务进程。cgroup处理窗口焦点的eventHandler回调函数

void CgroupEventHandler::HandleFocusedWindow(uint32_t windowId, uint64_t abilityToken,
    WindowType windowType, uint64_t displayId, int32_t pid, int32_t uid)
{
    ...
   
        supervisor_->focusedApp_ = app;
        SchedController::GetInstance().AdjustAllProcessGroup(*(app.get()), AdjustSource::ADJS_FOCUSED_WINDOW);
    }
    payload["bundleName"] = app->name_;
    ResSchedUtils::GetInstance().ReportDataInProcess(ResType::RES_TYPE_WINDOW_FOCUS, 0, payload);
}
调整进程组
void SchedController::AdjustAllProcessGroup(Application &app, AdjustSource source)
{
    ...
   
    cgAdjuster_->AdjustAllProcessGroup(app, source);
}

void CgroupAdjuster::AdjustProcessGroup(Application &app, ProcessRecord &pr, AdjustSource source)
{
    CGS_LOGI("%{public}s for %{public}d, source : %{public}d", __func__, pr.GetPid(), source);
    ComputeProcessGroup(app, pr, source);
    ApplyProcessGroup(app, pr);
}

void CgroupAdjuster::ComputeProcessGroup(Application &app, ProcessRecord &pr, AdjustSource source)
{
    SchedPolicy group = SchedPolicy::SP_DEFAULT;
   
    ...

        if (group == SchedPolicy::SP_BACKGROUND && pr.runningContinuousTask_) {
            group = SchedPolicy::SP_FOREGROUND; // move background key task to fg
        }

        pr.setSchedGroup_ = group;
    }
}

void CgroupAdjuster::ApplyProcessGroup(Application &app, ProcessRecord &pr)
{
    ChronoScope cs("ApplyProcessGroup");
    if (pr.curSchedGroup_ != pr.setSchedGroup_) {
        pid_t pid = pr.GetPid();
        int ret = CgroupSetting::SetThreadGroupSchedPolicy(pid, (int)pr.setSchedGroup_);
        if (ret != 0) {
            CGS_LOGE("%{public}s set %{public}d to group %{public}d failed, ret=%{public}d!",
                __func__, pid, pr.setSchedGroup_, ret);
            return;
        }
        
        ...
        
    }
}

通过进程间通讯IPC通知资源调度服务进程

void ResSchedClient::ReportDataInProcess(uint32_t resType, int64_t value, const Json::Value& payload)
{
    RESSCHED_LOGI("ResSchedClient::ReportDataInProcess receive resType = %{public}u, value = %{public}lld.",
        resType, value);
    ResSchedMgr::GetInstance().ReportData(resType, value, payload);
}

void ResSchedServiceProxy::ReportData(uint32_t resType, int64_t value, const Json::Value& payload)
{
    ...
   
    error = Remote()->SendRequest(IResSchedService::REPORT_DATA, data, reply, option);
    if (error != NO_ERROR) {
        RESSCHED_LOGE("Send request error: %{public}d", error);
        return;
    }
    RESSCHED_LOGD("ResSchedServiceProxy::ReportData success.");
}

int ResSchedServiceStub::OnRemoteRequest(uint32_t code, MessageParcel &data,
    MessageParcel &reply, MessageOption &option)
{
    auto uid = IPCSkeleton::GetCallingUid();
    RESSCHED_LOGD("ResSchedServiceStub::OnRemoteRequest, code = %{public}u, flags = %{public}d,"
        " uid = %{public}d", code, option.GetFlags(), uid);

    auto itFunc = funcMap_.find(code);
    if (itFunc != funcMap_.end()) {
        auto requestFunc = itFunc->second;
        if (requestFunc) {
            return requestFunc(data, reply);
        }
    }
    return IPCObjectStub::OnRemoteRequest(code, data, reply, option);
}