tests/eas/generic.py

# SPDX-License-Identifier: Apache-2.0
#
# Copyright (C) 2016, ARM Limited and contributors.
#
# Licensed 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.
#

from math import isnan

import numpy as np
import pandas as pd

from bart.common.Utils import area_under_curve

from energy_model import EnergyModel, EnergyModelCapacityError
from perf_analysis import PerfAnalysis
from test import LisaTest, experiment_test
from trace import Trace
from unittest import SkipTest


WORKLOAD_PERIOD_MS =  10
SET_IS_BIG_LITTLE = True
SET_INITIAL_TASK_UTIL = True

class _EnergyModelTest(LisaTest):
    """
    "Abstract" base class for generic EAS tests using the EnergyModel class

    Subclasses should provide a .workloads member to populate the 'wloads' field
    of the experiments_conf for the Executor. A set of helper methods are
    provided for making assertions about behaviour, most importantly the _test*
    methods which make assertions in a generic way.
    """

    test_conf = {
        "ftrace" : {
            "events" : [
                "sched_overutilized",
                "sched_energy_diff",
                "sched_load_avg_task",
                "sched_load_avg_cpu",
                "sched_migrate_task",
                "sched_switch",
                "cpu_frequency",
                "cpu_idle",
                "cpu_capacity",
            ],
        },
        "modules": ["cgroups"],
    }

    negative_slack_allowed_pct = 15
    """Percentage of RT-App task activations with negative slack allowed"""

    energy_est_threshold_pct = 20
    """
    Allowed margin for error in estimated energy cost for task placement,
    compared to optimal placment.
    """

    @classmethod
    def setUpClass(cls, *args, **kwargs):
        super(_EnergyModelTest, cls).runExperiments(*args, **kwargs)

    @classmethod
    def _getExperimentsConf(cls, test_env):
        if not test_env.nrg_model:
            try:
                test_env.nrg_model = EnergyModel.from_target(test_env.target)
            except Exception as e:
                raise SkipTest(
                    'This test requires an EnergyModel for the platform. '
                    'Either provide one manually or ensure it can be read '
                    'from the filesystem: {}'.format(e))

        conf = {
            'tag' : 'energy_aware',
            'flags' : ['ftrace', 'freeze_userspace'],
            'sched_features' : 'ENERGY_AWARE',
        }

        if 'cpufreq' in test_env.target.modules:
            available_govs = test_env.target.cpufreq.list_governors(0)
            if 'schedutil' in available_govs:
                conf['cpufreq'] = {'governor' : 'schedutil'}
            elif 'sched' in available_govs:
                conf['cpufreq'] = {'governor' : 'sched'}

        return {
            'wloads' : cls.workloads,
            'confs' : [conf],
        }

    @classmethod
    def _experimentsInit(cls, *args, **kwargs):
        super(_EnergyModelTest, cls)._experimentsInit(*args, **kwargs)

        if SET_IS_BIG_LITTLE:
            # This flag doesn't exist on mainline-integration kernels, so
            # don't worry if the file isn't present (hence verify=False)
            cls.target.write_value(
                "/proc/sys/kernel/sched_is_big_little", 1, verify=False)

        if SET_INITIAL_TASK_UTIL:
            # This flag doesn't exist on all kernels, so don't worry if the file
            # isn't present (hence verify=False)
            cls.target.write_value(
                "/proc/sys/kernel/sched_initial_task_util", 1024, verify=False)


    def get_task_utils_df(self, experiment):
        """
        Get a DataFrame with the *expected* utilization of each task over time

        :param experiment: The :class:Experiment to examine
        :returns: A Pandas DataFrame with a column for each task, showing how
                  the utilization of that task varies over time
        """
        util_scale = self.te.nrg_model.capacity_scale

        transitions = {}
        def add_transition(time, task, util):
            if time not in transitions:
                transitions[time] = {task: util}
            else:
                transitions[time][task] = util

        # First we'll build a dict D {time: {task_name: util}} where D[t][n] is
        # the expected utilization of task n from time t.
        for task, params in experiment.wload.params['profile'].iteritems():
            time = self.get_start_time(experiment) + params['delay']
            add_transition(time, task, 0)
            for _ in range(params.get('loops', 1)):
                for phase in params['phases']:
                    util = (phase.duty_cycle_pct * util_scale / 100.)
                    add_transition(time, task, util)
                    time += phase.duration_s
            add_transition(time, task, 0)

        index = sorted(transitions.keys())
        df = pd.DataFrame([transitions[k] for k in index], index=index)
        return df.fillna(method='ffill')

    def get_task_cpu_df(self, experiment):
        """
        Get a DataFrame mapping task names to the CPU they ran on

        Use the sched_switch trace event to find which CPU each task ran
        on. Does not reflect idleness - tasks not running are shown as running
        on the last CPU they woke on.

        :param experiment: The :class:Experiment to examine
        :returns: A Pandas DataFrame with a column for each task, showing the
                  CPU that the task was "on" at each moment in time
        """
        tasks = experiment.wload.tasks.keys()
        trace = self.get_trace(experiment)

        df = trace.ftrace.sched_switch.data_frame[['next_comm', '__cpu']]
        df = df[df['next_comm'].isin(tasks)]
        df = df.pivot(index=df.index, columns='next_comm').fillna(method='ffill')
        cpu_df = df['__cpu']
        # Drop consecutive duplicates
        cpu_df = cpu_df[(cpu_df.shift(+1) != cpu_df).any(axis=1)]
        return cpu_df

    def _sort_power_df_columns(self, df):
        """
        Helper method to re-order the columns of a power DataFrame

        This has no significance for code, but when examining DataFrames by hand
        they are easier to understand if the columns are in a logical order.
        """
        node_cpus = [node.cpus for node in self.te.nrg_model.root.iter_nodes()]
        return pd.DataFrame(df, columns=[c for c in node_cpus if c in df])

    def get_power_df(self, experiment):
        """
        Considering only the task placement, estimate power usage over time

        Examine a trace and use :meth:EnergyModel.estimate_from_cpu_util to get
        a DataFrame showing the estimated power usage over time. This assumes
        perfect cpuidle and cpufreq behaviour.

        :param experiment: The :class:Experiment to examine
        :returns: A Pandas DataFrame with a column node in the energy model
                  (keyed with a tuple of the CPUs contained by that node) Shows
                  the estimated power over time.
        """
        task_cpu_df = self.get_task_cpu_df(experiment)
        task_utils_df = self.get_task_utils_df(experiment)

        tasks = experiment.wload.tasks.keys()

        # Create a combined DataFrame with the utilization of a task and the CPU
        # it was running on at each moment. Looks like:
        #                       utils                  cpus
        #          task_wmig0 task_wmig1 task_wmig0 task_wmig1
        # 2.375056      102.4      102.4        NaN        NaN
        # 2.375105      102.4      102.4        2.0        NaN

        df = pd.concat([task_utils_df, task_cpu_df],
                       axis=1, keys=['utils', 'cpus'])
        df = df.sort_index().fillna(method='ffill')
        nrg_model = self.executor.te.nrg_model

        # Now make a DataFrame with the estimated power at each moment.
        def est_power(row):
            cpu_utils = [0 for cpu in nrg_model.cpus]
            for task in tasks:
                cpu = row['cpus'][task]
                util = row['utils'][task]
                if not isnan(cpu):
                    cpu_utils[int(cpu)] += util
            power = nrg_model.estimate_from_cpu_util(cpu_utils)
            columns = power.keys()
            return pd.Series([power[c] for c in columns], index=columns)
        return self._sort_power_df_columns(df.apply(est_power, axis=1))

    def get_expected_power_df(self, experiment):
        """
        Estimate *optimal* power usage over time

        Examine a trace and use :meth:get_optimal_placements and
        :meth:EnergyModel.estimate_from_cpu_util to get a DataFrame showing the
        estimated power usage over time under ideal EAS behaviour.

        :param experiment: The :class:Experiment to examine
        :returns: A Pandas DataFrame with a column each node in the energy model
                  (keyed with a tuple of the CPUs contained by that node) and a
                  "power" column with the sum of other columns. Shows the
                  estimated *optimal* power over time.
        """
        task_utils_df = self.get_task_utils_df(experiment)

        nrg_model = self.te.nrg_model

        def exp_power(row):
            task_utils = row.to_dict()
            expected_utils = nrg_model.get_optimal_placements(task_utils)
            power = nrg_model.estimate_from_cpu_util(expected_utils[0])
            columns = power.keys()
            return pd.Series([power[c] for c in columns], index=columns)
        return self._sort_power_df_columns(
            task_utils_df.apply(exp_power, axis=1))

    def _test_slack(self, experiment, tasks):
        """
        Assert that the RTApp workload was given enough performance

        Use :class:PerfAnalysis to find instances where the experiment's RT-App
        workload wasn't able to complete its activations (i.e. its reported
        "slack" was negative). Assert that this happened less that
        ``negative_slack_allowed_pct`` percent of the time.

        :meth:_test_task_placement asserts that estimated energy usage was
        low. That will pass for runs where too *little* energy was used,
        compromising performance. This method provides a separate test to
        counteract that problem.
        """

        pa = PerfAnalysis(experiment.out_dir)
        for task in tasks:
            slack = pa.df(task)["Slack"]

            bad_activations_pct = len(slack[slack < 0]) * 100. / len(slack)
            if bad_activations_pct > self.negative_slack_allowed_pct:
                raise AssertionError("task {} missed {}% of activations".format(
                    task, bad_activations_pct))

    def _test_task_placement(self, experiment, tasks):
        """
        Test that task placement was energy-efficient

        Use :meth:get_expected_power_df and :meth:get_power_df to estimate
        optimal and observed power usage for task placements of the experiment's
        workload. Assert that the observed power does not exceed the optimal
        power by more than 20%.
        """
        exp_power = self.get_expected_power_df(experiment)
        est_power = self.get_power_df(experiment)

        exp_energy = area_under_curve(exp_power.sum(axis=1), method='rect')
        est_energy = area_under_curve(est_power.sum(axis=1), method='rect')

        msg = 'Estimated {} bogo-Joules to run workload, expected {}'.format(
            est_energy, exp_energy)
        threshold = exp_energy * (1 + (self.energy_est_threshold_pct / 100.))
        self.assertLess(est_energy, threshold, msg=msg)

class OneSmallTask(_EnergyModelTest):
    """
    Test EAS for a single 20% task over 2 seconds
    """
    workloads = {
        'one_small' : {
            'type' : 'rt-app',
            'conf' : {
                'class' : 'periodic',
                'params' : {
                    'duty_cycle_pct': 20,
                    'duration_s': 2,
                    'period_ms': 10,
                },
                'tasks' : 1,
                'prefix' : 'many',
            },
        },
    }
    @experiment_test
    def test_slack(self, experiment, tasks):
        self._test_slack(experiment, tasks)
    @experiment_test
    def test_task_placement(self, experiment, tasks):
        self._test_task_placement(experiment, tasks)

class ThreeSmallTasks(_EnergyModelTest):
    """
    Test EAS for 3 20% tasks over 2 seconds
    """
    workloads = {
        'three_small' : {
            'type' : 'rt-app',
            'conf' : {
                'class' : 'periodic',
                'params' : {
                    'duty_cycle_pct': 20,
                    'duration_s': 2,
                    'period_ms': 10,
                },
                'tasks' : 3,
                'prefix' : 'many',
            },
        },
    }
    @experiment_test
    def test_slack(self, experiment, tasks):
        self._test_slack(experiment, tasks)
    @experiment_test
    def test_task_placement(self, experiment, tasks):
        self._test_task_placement(experiment, tasks)

class TwoBigTasks(_EnergyModelTest):
    """
    Test EAS for 2 80% tasks over 2 seconds
    """
    workloads = {
        'two_big' : {
            'type' : 'rt-app',
            'conf' : {
                'class' : 'periodic',
                'params' : {
                    'duty_cycle_pct': 80,
                    'duration_s': 2,
                    'period_ms': 10,
                },
                'tasks' : 2,
                'prefix' : 'many',
            },
        },
    }
    @experiment_test
    def test_slack(self, experiment, tasks):
        self._test_slack(experiment, tasks)
    @experiment_test
    def test_task_placement(self, experiment, tasks):
        self._test_task_placement(experiment, tasks)

class TwoBigThreeSmall(_EnergyModelTest):
    """
    Test EAS for 2 70% tasks and 3 10% tasks over 2 seconds
    """
    workloads = {
        'two_big_three_small' : {
            'type' : 'rt-app',
            'conf' : {
                'class' : 'profile',
                'params' : {
                    'large' : {
                        'kind' : 'Periodic',
                        'params' : {
                            'duty_cycle_pct': 70,
                            'duration_s': 2,
                            'period_ms': WORKLOAD_PERIOD_MS,
                        },
                        'tasks' : 2,
                    },
                    'small' : {
                        'kind' : 'Periodic',
                        'params' : {
                            'duty_cycle_pct': 10,
                            'duration_s': 2,
                            'period_ms': WORKLOAD_PERIOD_MS,
                        },
                        'tasks' : 3,
                    },
                },
            },
        },
    }
    @experiment_test
    def test_slack(self, experiment, tasks):
        self._test_slack(experiment, tasks)
    @experiment_test
    def test_task_placement(self, experiment, tasks):
        self._test_task_placement(experiment, tasks)

class RampUp(_EnergyModelTest):
    """
    Test EAS for a task ramping from 5% up to 70% over 2 seconds
    """
    workloads = {
        "ramp_up" : {
            "type": "rt-app",
            "conf" : {
                "class"  : "profile",
                "params"  : {
                    "r5_10-60" : {
                        "kind"   : "Ramp",
                        "params" : {
                            "period_ms" : 16,
                            "start_pct" :  5,
                            "end_pct"   : 70,
                            "delta_pct" :  5,
                            "time_s"    :  2,
                         },
                    },
                },
            },
        },
    }

    @experiment_test
    def test_slack(self, experiment, tasks):
        self._test_slack(experiment, tasks)
    @experiment_test
    def test_task_placement(self, experiment, tasks):
        self._test_task_placement(experiment, tasks)

class RampDown(_EnergyModelTest):
    """
    Test EAS for a task ramping from 70% down to 5% over 2 seconds
    """
    workloads = {
        "ramp_down" : {
            "type": "rt-app",
            "conf" : {
                "class"  : "profile",
                "params"  : {
                    "r5_10-60" : {
                        "kind"   : "Ramp",
                        "params" : {
                            "period_ms" : 16,
                            "start_pct" : 70,
                            "end_pct"   :  5,
                            "delta_pct" :  5,
                            "time_s"    :  2,
                         },
                    },
                },
            },
        },
    }

    @experiment_test
    def test_slack(self, experiment, tasks):
        self._test_slack(experiment, tasks)
    @experiment_test
    def test_task_placement(self, experiment, tasks):
        self._test_task_placement(experiment, tasks)

class EnergyModelWakeMigration(_EnergyModelTest):
    """
    Test EAS for tasks alternating beetween 10% and 50%
    """
    workloads = {
        'em_wake_migration' : {
            'type' : 'rt-app',
            'conf' : {
                'class' : 'profile',
                'params' : {
                    'wmig' : {
                        'kind' : 'Step',
                        'params' : {
                            'start_pct': 10,
                            'end_pct': 50,
                            'time_s': 2,
                            'loops': 2
                        },
                        # Create one task for each big cpu
                        'tasks' : 'big',
                    },
                },
            },
        },
    }
    @experiment_test
    def test_slack(self, experiment, tasks):
        self._test_slack(experiment, tasks)
    @experiment_test
    def test_task_placement(self, experiment, tasks):
        self._test_task_placement(experiment, tasks)