wlgen/wlgen/rta.py

# SPDX-License-Identifier: Apache-2.0
#
# Copyright (C) 2015, 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.
#

import fileinput
import json
import os
import re

from collections import namedtuple
from wlgen import Workload
from devlib.utils.misc import ranges_to_list

import logging

_Phase = namedtuple('Phase', 'duration_s, period_ms, duty_cycle_pct')
class Phase(_Phase):
    """
    Descriptor for an RT-App load phase

    :param duration_s: the phase duration in [s].
    :type duration_s: int

    :param period_ms: the phase period in [ms].
    :type period_ms: int

    :param duty_cycle_pct: the generated load in [%].
    :type duty_cycle_pct: int
    """
    pass

class RTA(Workload):
    """
    Class for creating RT-App workloads
    """

    def __init__(self,
                 target,
                 name,
                 calibration=None):
        """
        :param target: Devlib target to run workload on.
        :param name: Human-readable name for the workload.
        :param calibration: CPU calibration specification. Can be obtained from
                            :meth:`calibrate`.
        """

        # Setup logging
        self._log = logging.getLogger('RTApp')

        # rt-app calibration
        self.pload = calibration

        # TODO: Assume rt-app is pre-installed on target
        # self.target.setup('rt-app')

        super(RTA, self).__init__(target, name)

        # rt-app executor
        self.wtype = 'rtapp'
        self.executor = 'rt-app'

        # Default initialization
        self.json = None
        self.rta_profile = None
        self.loadref = None
        self.rta_cmd  = None
        self.rta_conf = None
        self.test_label = None

        # Setup RTA callbacks
        self.setCallback('postrun', self.__postrun)

    @staticmethod
    def calibrate(target):
        """
        Calibrate RT-App on each CPU in the system

        :param target: Devlib target to run calibration on.
        :returns: Dict mapping CPU numbers to RT-App calibration values.
        """
        pload_regexp = re.compile(r'pLoad = ([0-9]+)ns')
        pload = {}

        # Setup logging
        log = logging.getLogger('RTApp')

        # Save previous governors
        old_governors = {}
        for domain in target.cpufreq.iter_domains():
            cpu = domain[0]
            governor = target.cpufreq.get_governor(cpu)
            tunables = target.cpufreq.get_governor_tunables(cpu)
            old_governors[cpu] = governor, tunables

        target.cpufreq.set_all_governors('performance')

        for cpu in target.list_online_cpus():

            log.info('CPU%d calibration...', cpu)

            max_rtprio = int(target.execute('ulimit -Hr').split('\r')[0])
            log.debug('Max RT prio: %d', max_rtprio)
            if max_rtprio > 10:
                max_rtprio = 10

            rta = RTA(target, 'rta_calib')
            rta.conf(kind='profile',
                    params = {
                        'task1': Periodic(
                            period_ms=100,
                            duty_cycle_pct=50,
                            duration_s=1,
                            sched={
                                'policy': 'FIFO',
                                'prio' : max_rtprio
                            }
                        ).get()
                    },
                    cpus=[cpu])
            rta.run(as_root=True)

            for line in rta.getOutput().split('\n'):
                pload_match = re.search(pload_regexp, line)
                if pload_match is None:
                    continue
                pload[cpu] = int(pload_match.group(1))
                log.debug('>>> cpu%d: %d', cpu, pload[cpu])

        # Restore previous governors
        #   Setting a governor & tunables for a cpu will set them for all cpus
        #   in the same clock domain, so only restoring them for one cpu
        #   per domain is enough to restore them all.
        for cpu, (governor, tunables) in old_governors.iteritems():
            target.cpufreq.set_governor(cpu, governor)
            target.cpufreq.set_governor_tunables(cpu, **tunables)

        log.info('Target RT-App calibration:')
        log.info("{" + ", ".join('"%r": %r' % (key, pload[key])
                                 for key in pload) + "}")

        # Sanity check calibration values for big.LITTLE systems
        if 'bl' in target.modules:
            bcpu = target.bl.bigs_online[0]
            lcpu = target.bl.littles_online[0]
            if pload[bcpu] > pload[lcpu]:
                log.warning('Calibration values reports big cores less '
                            'capable than LITTLE cores')
                raise RuntimeError('Calibration failed: try again or file a bug')
            bigs_speedup = ((float(pload[lcpu]) / pload[bcpu]) - 1) * 100
            log.info('big cores are ~%.0f%% more capable than LITTLE cores',
                     bigs_speedup)

        return pload

    def __postrun(self, params):
        destdir = params['destdir']
        if destdir is None:
            return
        self._log.debug('Pulling logfiles to [%s]...', destdir)
        for task in self.tasks.keys():
            logfile = self.target.path.join(self.run_dir,
                                            '*{}*.log'.format(task))
            self.target.pull(logfile, destdir)
        self._log.debug('Pulling JSON to [%s]...', destdir)
        self.target.pull(self.target.path.join(self.run_dir, self.json),
                         destdir)
        logfile = self.target.path.join(destdir, 'output.log')
        self._log.debug('Saving output on [%s]...', logfile)
        with open(logfile, 'w') as ofile:
            for line in self.output['executor'].split('\n'):
                ofile.write(line+'\n')

    def getCalibrationConf(self):
        # Select CPU for task calibration, which is the first little
        # of big depending on the loadref tag
        if self.pload is not None:
            if self.loadref and self.loadref.upper() == 'LITTLE':
                return max(self.pload.values())
            else:
                return min(self.pload.values())
        else:
            cpus = self.cpus or range(self.target.number_of_cpus)

            target_cpu = cpus[-1]
            if 'bl'in self.target.modules:
                cluster = self.target.bl.bigs
                candidates = sorted(set(self.target.bl.bigs).intersection(cpus))
                if candidates:
                    target_cpu = candidates[0]

            return 'CPU{0:d}'.format(target_cpu)

    def _confCustom(self):

        rtapp_conf = self.params['custom']

        # Sanity check params being a valid file path
        if not isinstance(rtapp_conf, str) or \
           not os.path.isfile(rtapp_conf):
            self._log.debug('Checking for %s', rtapp_conf)
            raise ValueError('value specified for \'params\' is not '
                             'a valid rt-app JSON configuration file')

        self._log.info('Loading custom configuration:')
        self._log.info('   %s', rtapp_conf)
        self.json = '{0:s}_{1:02d}.json'.format(self.name, self.exc_id)
        ofile = open(self.json, 'w')
        ifile = open(rtapp_conf, 'r')

        calibration = self.getCalibrationConf()
        # Calibration can either be a string like "CPU1" or an integer, if the
        # former we need to quote it.
        if type(calibration) != int:
            calibration = '"{}"'.format(calibration)

        replacements = {
            '__DURATION__' : str(self.duration),
            '__PVALUE__'   : str(calibration),
            '__LOGDIR__'   : str(self.run_dir),
            '__WORKDIR__'  : '"'+self.target.working_directory+'"',
        }

        for line in ifile:
            if '__DURATION__' in line and self.duration is None:
                raise ValueError('Workload duration not specified')
            for src, target in replacements.iteritems():
                line = line.replace(src, target)
            ofile.write(line)
        ifile.close()
        ofile.close()

        with open(self.json) as f:
            conf = json.load(f)
        for tid in conf['tasks']:
            self.tasks[tid] = {'pid': -1}

        return self.json

    def _confProfile(self):

        # Sanity check for task names
        for task in self.params['profile'].keys():
            if len(task) > 15:
                # rt-app uses pthread_setname_np(3) which limits the task name
                # to 16 characters including the terminal '\0'.
                msg = ('Task name "{}" too long, please configure your tasks '
                       'with names shorter than 16 characters').format(task)
                raise ValueError(msg)

        # Task configuration
        self.rta_profile = {
            'tasks': {},
            'global': {}
        }

        # Initialize global configuration
        global_conf = {
                'default_policy': 'SCHED_OTHER',
                'duration': -1,
                'calibration': self.getCalibrationConf(),
                'logdir': self.run_dir,
            }

        if self.duration is not None:
            global_conf['duration'] = self.duration
            self._log.warn('Limiting workload duration to %d [s]',
                           global_conf['duration'])
        else:
            self._log.info('Workload duration defined by longest task')

        # Setup default scheduling class
        if 'policy' in self.sched:
            policy = self.sched['policy'].upper()
            if policy not in ['OTHER', 'FIFO', 'RR', 'DEADLINE']:
                raise ValueError('scheduling class {} not supported'\
                        .format(policy))
            global_conf['default_policy'] = 'SCHED_' + self.sched['policy']

        self._log.info('Default policy: %s', global_conf['default_policy'])

        # Setup global configuration
        self.rta_profile['global'] = global_conf

        # Setup tasks parameters
        for tid in sorted(self.params['profile'].keys()):
            task = self.params['profile'][tid]

            # Initialize task configuration
            task_conf = {}

            if 'sched' not in task:
                policy = 'DEFAULT'
            else:
                policy = task['sched']['policy'].upper()
            if policy == 'DEFAULT':
                task_conf['policy'] = global_conf['default_policy']
                sched_descr = 'sched: using default policy'
            elif policy not in ['OTHER', 'FIFO', 'RR', 'DEADLINE']:
                raise ValueError('scheduling class {} not supported'\
                        .format(task['sclass']))
            else:
                task_conf.update(task['sched'])
                task_conf['policy'] = 'SCHED_' + policy
                sched_descr = 'sched: {0:s}'.format(task['sched'])

            # Initialize task phases
            task_conf['phases'] = {}

            self._log.info('------------------------')
            self._log.info('task [%s], %s', tid, sched_descr)

            if 'delay' in task.keys():
                if task['delay'] > 0:
                    task_conf['delay'] = int(task['delay'] * 1e6)
                    self._log.info(' | start delay: %.6f [s]',
                            task['delay'])

            if 'loops' not in task.keys():
                task['loops'] = 1
            task_conf['loop'] = task['loops']
            self._log.info(' | loops count: %d', task['loops'])

            # Setup task affinity
            if 'cpus' in task and task['cpus']:
                self._log.info(' | CPUs affinity: %s', task['cpus'])
                if isinstance(task['cpus'], str):
                    task_conf['cpus'] = ranges_to_list(task['cpus'])
                elif isinstance(task['cpus'], list):
                    task_conf['cpus'] = task['cpus']
                else:
                    raise ValueError('cpus must be a list or string')


            # Setup task configuration
            self.rta_profile['tasks'][tid] = task_conf

            # Getting task phase descriptor
            pid=1
            for phase in task['phases']:

                # Convert time parameters to integer [us] units
                duration = int(phase.duration_s * 1e6)
                period = int(phase.period_ms * 1e3)

                # A duty-cycle of 0[%] translates on a 'sleep' phase
                if phase.duty_cycle_pct == 0:

                    self._log.info(' + phase_%06d: sleep %.6f [s]',
                                   pid, duration/1e6)

                    task_phase = {
                        'loop': 1,
                        'sleep': duration,
                    }

                # A duty-cycle of 100[%] translates on a 'run-only' phase
                elif phase.duty_cycle_pct == 100:

                    self._log.info(' + phase_%06d: batch %.6f [s]',
                                   pid, duration/1e6)

                    task_phase = {
                        'loop': 1,
                        'run': duration,
                    }

                # A certain number of loops is requires to generate the
                # proper load
                else:

                    cloops = -1
                    if duration >= 0:
                        cloops = int(duration / period)

                    sleep_time = period * (100 - phase.duty_cycle_pct) / 100
                    running_time = period - sleep_time

                    self._log.info('+ phase_%06d: duration %.6f [s] (%d loops)',
                                   pid, duration/1e6, cloops)
                    self._log.info('|  period   %6d [us], duty_cycle %3d %%',
                                   period, phase.duty_cycle_pct)
                    self._log.info('|  run_time %6d [us], sleep_time %6d [us]',
                                   running_time, sleep_time)

                    task_phase = {
                        'loop': cloops,
                        'run': running_time,
                        'timer': {'ref': tid, 'period': period},
                    }

                self.rta_profile['tasks'][tid]['phases']\
                    ['p'+str(pid).zfill(6)] = task_phase

                pid+=1

            # Append task name to the list of this workload tasks
            self.tasks[tid] = {'pid': -1}

        # Generate JSON configuration on local file
        self.json = '{0:s}_{1:02d}.json'.format(self.name, self.exc_id)
        with open(self.json, 'w') as outfile:
            json.dump(self.rta_profile, outfile,
                    sort_keys=True, indent=4, separators=(',', ': '))

        return self.json

    def conf(self,
             kind,
             params,
             duration=None,
             cpus=None,
             sched=None,
             run_dir=None,
             exc_id=0,
             loadref='big'):
        """
        Configure a workload of a specified kind.

        The rt-app based workload allows to define different classes of
        workloads. The classes supported so far are detailed hereafter.

        Custom workloads
          When 'kind' is 'custom' the tasks generated by this workload are the
          ones defined in a provided rt-app JSON configuration file.
          In this case the 'params' parameter must be used to specify the
          complete path of the rt-app JSON configuration file to use.

        Profile based workloads
          When ``kind`` is "profile", ``params`` is a dictionary mapping task
          names to task specifications. The easiest way to create these task
          specifications using :meth:`RTATask.get`.

          For example, the following configures an RTA workload with a single
          task, named 't1', using the default parameters for a Periodic RTATask:

          ::

            wl = RTA(...)
            wl.conf(kind='profile', params={'t1': Periodic().get()})

        :param kind: Either 'custom' or 'profile' - see above.
        :param params: RT-App parameters - see above.
        :param duration: Maximum duration of the workload in seconds. Any
                         remaining tasks are killed by rt-app when this time has
                         elapsed.
        :param cpus: CPUs to restrict this workload to, using ``taskset``.
        :type cpus: list(int)

        :param sched: Global RT-App scheduler configuration. Dict with fields:

          policy
            The default scheduler policy. Choose from 'OTHER', 'FIFO', 'RR',
            and 'DEADLINE'.

        :param run_dir: Target dir to store output and config files in.

        .. TODO: document or remove loadref
        """

        if not sched:
            sched = {'policy' : 'OTHER'}

        super(RTA, self).conf(kind, params, duration,
                cpus, sched, run_dir, exc_id)

        self.loadref = loadref

        # Setup class-specific configuration
        if kind == 'custom':
            self._confCustom()
        elif kind == 'profile':
            self._confProfile()

        # Move configuration file to target
        self.target.push(self.json, self.run_dir)

        self.rta_cmd  = self.target.executables_directory + '/rt-app'
        self.rta_conf = self.run_dir + '/' + self.json
        self.command = '{0:s} {1:s} 2>&1'.format(self.rta_cmd, self.rta_conf)

        # Set and return the test label
        self.test_label = '{0:s}_{1:02d}'.format(self.name, self.exc_id)
        return self.test_label

class RTATask(object):
    """
    Base class for conveniently constructing params to :meth:`RTA.conf`

    This class represents an RT-App task which may contain multiple phases. It
    implements ``__add__`` so that using ``+`` on two tasks concatenates their
    phases. For example ``Ramp() + Periodic()`` would yield an ``RTATask`` that
    executes the default phases for ``Ramp`` followed by the default phases for
    ``Periodic``.
    """

    def __init__(self):
        self._task = {}

    def get(self):
        """
        Return a dict that can be passed as an element of the ``params`` field
        to :meth:`RTA.conf`.
        """
        return self._task

    def __add__(self, next_phases):
        if next_phases._task.get('delay', 0):
            # This won't work, because rt-app's "delay" field is per-task and
            # not per-phase. We might be able to implement it by adding a
            # "sleep" event here, but let's not bother unless such a need
            # arises.
            raise ValueError("Can't compose rt-app tasks "
                             "when the second has nonzero 'delay_s'")

        self._task['phases'].extend(next_phases._task['phases'])
        return self


class Ramp(RTATask):
    """
    Configure a ramp load.

    This class defines a task which load is a ramp with a configured number
    of steps according to the input parameters.

    :param start_pct: the initial load percentage.
    :param end_pct: the final load percentage.
    :param delta_pct: the load increase/decrease at each step, in percentage
                      points.
    :param time_s: the duration in seconds of each load step.
    :param period_ms: the period used to define the load in [ms].
    :param delay_s: the delay in seconds before ramp start.
    :param loops: number of time to repeat the ramp, with the specified delay in
                  between.

    :param sched: the scheduler configuration for this task.
    :type sched: dict

    :param cpus: the list of CPUs on which task can run.
    :type cpus: list(int)
    """

    def __init__(self, start_pct=0, end_pct=100, delta_pct=10, time_s=1,
                 period_ms=100, delay_s=0, loops=1, sched=None, cpus=None):
        super(Ramp, self).__init__()

        self._task['cpus'] = cpus
        if not sched:
            sched = {'policy' : 'DEFAULT'}
        self._task['sched'] = sched
        self._task['delay'] = delay_s
        self._task['loops'] = loops

        if start_pct not in range(0,101) or end_pct not in range(0,101):
            raise ValueError('start_pct and end_pct must be in [0..100] range')

        if start_pct >= end_pct:
            if delta_pct > 0:
                delta_pct = -delta_pct
            delta_adj = -1
        if start_pct <= end_pct:
            if delta_pct < 0:
                delta_pct = -delta_pct
            delta_adj = +1

        phases = []
        steps = range(start_pct, end_pct+delta_adj, delta_pct)
        for load in steps:
            if load == 0:
                phase = Phase(time_s, 0, 0)
            else:
                phase = Phase(time_s, period_ms, load)
            phases.append(phase)

        self._task['phases'] = phases

class Step(Ramp):
    """
    Configure a step load.

    This class defines a task which load is a step with a configured initial and
    final load. Using the ``loops`` param, this can be used to create a workload
    that alternates between two load values.

    :param start_pct: the initial load percentage.
    :param end_pct: the final load percentage.
    :param time_s: the duration in seconds of each load step.
    :param period_ms: the period used to define the load in [ms].
    :param delay_s: the delay in seconds before ramp start.
    :param loops: number of time to repeat the step, with the specified delay in
                  between.

    :param sched: the scheduler configuration for this task.
    :type sched: dict

    :param cpus: the list of CPUs on which task can run.
    :type cpus: list(int)
    """

    def __init__(self, start_pct=0, end_pct=100, time_s=1, period_ms=100,
                 delay_s=0, loops=1, sched=None, cpus=None):
        delta_pct = abs(end_pct - start_pct)
        super(Step, self).__init__(start_pct, end_pct, delta_pct, time_s,
                                   period_ms, delay_s, loops, sched, cpus)

class Pulse(RTATask):
    """
    Configure a pulse load.

    This class defines a task which load is a pulse with a configured
    initial and final load.

    The main difference with the 'step' class is that a pulse workload is
    by definition a 'step down', i.e. the workload switch from an finial
    load to a final one which is always lower than the initial one.
    Moreover, a pulse load does not generate a sleep phase in case of 0[%]
    load, i.e. the task ends as soon as the non null initial load has
    completed.

    :param start_pct: the initial load percentage.
    :param end_pct: the final load percentage. Must be lower than ``start_pct``
                    value. If end_pct is 0, the task end after the ``start_pct``
                    period has completed.
    :param time_s: the duration in seconds of each load step.
    :param period_ms: the period used to define the load in [ms].
    :param delay_s: the delay in seconds before ramp start.
    :param loops: number of time to repeat the pulse, with the specified delay
                  in between.

    :param sched: the scheduler configuration for this task.
    :type sched: dict

    :param cpus: the list of CPUs on which task can run
    :type cpus: list(int)
    """

    def __init__(self, start_pct=100, end_pct=0, time_s=1, period_ms=100,
                 delay_s=0, loops=1, sched=None, cpus=None):
        super(Pulse, self).__init__()

        if end_pct >= start_pct:
            raise ValueError('end_pct must be lower than start_pct')

        self._task = {}

        self._task['cpus'] = cpus
        if not sched:
            sched = {'policy' : 'DEFAULT'}
        self._task['sched'] = sched
        self._task['delay'] = delay_s
        self._task['loops'] = loops
        self._task['phases'] = {}

        if end_pct not in range(0,101) or start_pct not in range(0,101):
            raise ValueError('end_pct and start_pct must be in [0..100] range')
        if end_pct >= start_pct:
            raise ValueError('end_pct must be lower than start_pct')

        phases = []
        for load in [start_pct, end_pct]:
            if load == 0:
                continue
            phase = Phase(time_s, period_ms, load)
            phases.append(phase)

        self._task['phases'] = phases


class Periodic(Pulse):
    """
    Configure a periodic load. This is the simplest type of RTA task.

    This class defines a task which load is periodic with a configured
    period and duty-cycle.

    :param duty_cycle_pct: the load percentage.
    :param duration_s: the total duration in seconds of the task.
    :param period_ms: the period used to define the load in milliseconds.
    :param delay_s: the delay in seconds before starting the periodic phase.

    :param sched: the scheduler configuration for this task.
    :type sched: dict

    :param cpus: the list of CPUs on which task can run.
    :type cpus: list(int)
    """

    def __init__(self, duty_cycle_pct=50, duration_s=1, period_ms=100,
                 delay_s=0, sched=None, cpus=None):
        super(Periodic, self).__init__(duty_cycle_pct, 0, duration_s,
                                       period_ms, delay_s, 1, sched, cpus)