| # 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. |
| # |
| |
| """ Trace Parser Module """ |
| |
| import numpy as np |
| import os |
| import pandas as pd |
| import sys |
| import trappy |
| import json |
| import warnings |
| import operator |
| import logging |
| |
| from analysis_register import AnalysisRegister |
| from collections import namedtuple |
| from devlib.utils.misc import memoized |
| from trappy.utils import listify, handle_duplicate_index |
| |
| |
| NON_IDLE_STATE = -1 |
| ResidencyTime = namedtuple('ResidencyTime', ['total', 'active']) |
| ResidencyData = namedtuple('ResidencyData', ['label', 'residency']) |
| |
| class Trace(object): |
| """ |
| The Trace object is the LISA trace events parser. |
| |
| :param platform: a dictionary containing information about the target |
| platform |
| :type platform: dict |
| |
| :param data_dir: folder containing all trace data |
| :type data_dir: str |
| |
| :param events: events to be parsed (everything in the trace by default) |
| :type events: list(str) |
| |
| :param tasks: filter data for the specified tasks only. If None (default), |
| use data for all tasks found in the trace. |
| :type tasks: list(str) or NoneType |
| |
| :param window: time window to consider when parsing the trace |
| :type window: tuple(int, int) |
| |
| :param normalize_time: normalize trace time stamps |
| :type normalize_time: bool |
| |
| :param trace_format: format of the trace. Possible values are: |
| - FTrace |
| - SysTrace |
| :type trace_format: str |
| |
| :param plots_dir: directory where to save plots |
| :type plots_dir: str |
| |
| :param plots_prefix: prefix for plots file names |
| :type plots_prefix: str |
| |
| :param cgroup_info: add cgroup information for sanitization |
| example: |
| { |
| 'controller_ids': { 2: 'schedtune', 4: 'cpuset' }, |
| 'cgroups': [ 'root', 'background', 'foreground' ], # list of allowed cgroup names |
| } |
| :type cgroup_info: dict |
| """ |
| |
| def __init__(self, platform, data_dir, events=None, |
| tasks=None, window=(0, None), |
| normalize_time=True, |
| trace_format='FTrace', |
| plots_dir=None, |
| plots_prefix='', |
| cgroup_info={}): |
| |
| # The platform used to run the experiments |
| self.platform = platform |
| |
| # TRAPpy Trace object |
| self.ftrace = None |
| |
| # Trace format |
| self.trace_format = trace_format |
| |
| # The time window used to limit trace parsing to |
| self.window = window |
| |
| # Dynamically registered TRAPpy events |
| self.trappy_cls = {} |
| |
| # Maximum timespan for all collected events |
| self.time_range = 0 |
| |
| # Time the system was overutilzied |
| self.overutilized_time = 0 |
| self.overutilized_prc = 0 |
| |
| # The dictionary of tasks descriptors available in the dataset |
| self.tasks = {} |
| |
| # List of events required by user |
| self.events = [] |
| |
| # List of events available in the parsed trace |
| self.available_events = [] |
| |
| # Cluster frequency coherency flag |
| self.freq_coherency = True |
| |
| # Folder containing all trace data |
| self.data_dir = None |
| |
| # Setup logging |
| self._log = logging.getLogger('Trace') |
| |
| # Folder containing trace |
| if not os.path.isdir(data_dir): |
| self.data_dir = os.path.dirname(data_dir) |
| else: |
| self.data_dir = data_dir |
| |
| # By deafult, use the trace dir to save plots |
| self.plots_dir = plots_dir |
| if self.plots_dir is None: |
| self.plots_dir = self.data_dir |
| self.plots_prefix = plots_prefix |
| |
| # Cgroup info for sanitization |
| self.cgroup_info = cgroup_info |
| |
| self.__registerTraceEvents(events) if events else None |
| self.__parseTrace(data_dir, tasks, window, normalize_time, |
| trace_format) |
| self.__computeTimeSpan() |
| |
| # Minimum and Maximum x_time to use for all plots |
| self.x_min = 0 |
| self.x_max = self.time_range |
| |
| # Reset x axis time range to full scale |
| t_min = self.window[0] |
| t_max = self.window[1] |
| self.setXTimeRange(t_min, t_max) |
| |
| self.data_frame = TraceData() |
| self._registerDataFrameGetters(self) |
| |
| self.analysis = AnalysisRegister(self) |
| |
| def _registerDataFrameGetters(self, module): |
| """ |
| Internal utility function that looks up getter functions with a "_dfg_" |
| prefix in their name and bounds them to the specified module. |
| |
| :param module: module to which the function is added |
| :type module: class |
| """ |
| self._log.debug('Registering [%s] local data frames', module) |
| for func in dir(module): |
| if not func.startswith('_dfg_'): |
| continue |
| dfg_name = func.replace('_dfg_', '') |
| dfg_func = getattr(module, func) |
| self._log.debug(' %s', dfg_name) |
| setattr(self.data_frame, dfg_name, dfg_func) |
| |
| def setXTimeRange(self, t_min=None, t_max=None): |
| """ |
| Set x axis time range to the specified values. |
| |
| :param t_min: lower bound |
| :type t_min: int or float |
| |
| :param t_max: upper bound |
| :type t_max: int or float |
| """ |
| if t_min is None: |
| self.x_min = 0 |
| else: |
| self.x_min = t_min |
| if t_max is None: |
| self.x_max = self.time_range |
| else: |
| self.x_max = t_max |
| self._log.debug('Set plots time range to (%.6f, %.6f)[s]', |
| self.x_min, self.x_max) |
| |
| def __registerTraceEvents(self, events): |
| """ |
| Save a copy of the parsed events. |
| |
| :param events: single event name or list of events names |
| :type events: str or list(str) |
| """ |
| if isinstance(events, basestring): |
| self.events = events.split(' ') |
| elif isinstance(events, list): |
| self.events = events |
| else: |
| raise ValueError('Events must be a string or a list of strings') |
| # Register devlib fake cpu_frequency events |
| if 'cpu_frequency' in events: |
| self.events.append('cpu_frequency_devlib') |
| |
| def __parseTrace(self, path, tasks, window, normalize_time, trace_format): |
| """ |
| Internal method in charge of performing the actual parsing of the |
| trace. |
| |
| :param path: path to the trace folder (or trace file) |
| :type path: str |
| |
| :param tasks: filter data for the specified tasks only |
| :type tasks: list(str) |
| |
| :param window: time window to consider when parsing the trace |
| :type window: tuple(int, int) |
| |
| :param normalize_time: normalize trace time stamps |
| :type normalize_time: bool |
| |
| :param trace_format: format of the trace. Possible values are: |
| - FTrace |
| - SysTrace |
| :type trace_format: str |
| """ |
| self._log.debug('Loading [sched] events from trace in [%s]...', path) |
| self._log.debug('Parsing events: %s', self.events if self.events else 'ALL') |
| if trace_format.upper() == 'SYSTRACE' or path.endswith('html'): |
| self._log.debug('Parsing SysTrace format...') |
| trace_class = trappy.SysTrace |
| self.trace_format = 'SysTrace' |
| elif trace_format.upper() == 'FTRACE': |
| self._log.debug('Parsing FTrace format...') |
| trace_class = trappy.FTrace |
| self.trace_format = 'FTrace' |
| else: |
| raise ValueError("Unknown trace format {}".format(trace_format)) |
| |
| scope = 'custom' if self.events else 'all' |
| self.ftrace = trace_class(path, scope=scope, events=self.events, |
| window=window, normalize_time=normalize_time) |
| |
| # Load Functions profiling data |
| has_function_stats = self._loadFunctionsStats(path) |
| |
| # Check for events available on the parsed trace |
| self.__checkAvailableEvents() |
| if len(self.available_events) == 0: |
| if has_function_stats: |
| self._log.info('Trace contains only functions stats') |
| return |
| raise ValueError('The trace does not contain useful events ' |
| 'nor function stats') |
| |
| # Sanitize cgroup info if any |
| self._sanitize_CgroupAttachTask() |
| |
| # Santization not possible if platform missing |
| if not self.platform: |
| # Setup internal data reference to interesting events/dataframes |
| self._sanitize_SchedLoadAvgCpu() |
| self._sanitize_SchedLoadAvgTask() |
| self._sanitize_SchedCpuCapacity() |
| self._sanitize_SchedBoostCpu() |
| self._sanitize_SchedBoostTask() |
| self._sanitize_SchedEnergyDiff() |
| self._sanitize_SchedOverutilized() |
| self._sanitize_CpuFrequency() |
| |
| self.__loadTasksNames(tasks) |
| |
| # Compute plot window |
| if not normalize_time: |
| start = self.window[0] |
| if self.window[1]: |
| duration = min(self.ftrace.get_duration(), self.window[1]) |
| else: |
| duration = self.ftrace.get_duration() |
| self.window = (self.ftrace.basetime + start, |
| self.ftrace.basetime + duration) |
| |
| def __checkAvailableEvents(self, key=""): |
| """ |
| Internal method used to build a list of available events. |
| |
| :param key: key to be used for TRAPpy filtering |
| :type key: str |
| """ |
| for val in self.ftrace.get_filters(key): |
| obj = getattr(self.ftrace, val) |
| if len(obj.data_frame): |
| self.available_events.append(val) |
| self._log.debug('Events found on trace:') |
| for evt in self.available_events: |
| self._log.debug(' - %s', evt) |
| |
| def __loadTasksNames(self, tasks): |
| """ |
| Try to load tasks names using one of the supported events. |
| |
| :param tasks: list of task names. If None, load all tasks found. |
| :type tasks: list(str) or NoneType |
| """ |
| def load(tasks, event, name_key, pid_key): |
| df = self._dfg_trace_event(event) |
| if tasks is None: |
| tasks = df[name_key].unique() |
| self.getTasks(df, tasks, name_key=name_key, pid_key=pid_key) |
| self._scanTasks(df, name_key=name_key, pid_key=pid_key) |
| self._scanTgids(df) |
| |
| if 'sched_switch' in self.available_events: |
| load(tasks, 'sched_switch', 'next_comm', 'next_pid') |
| elif 'sched_load_avg_task' in self.available_events: |
| load(tasks, 'sched_load_avg_task', 'comm', 'pid') |
| else: |
| self._log.warning('Failed to load tasks names from trace events') |
| |
| def hasEvents(self, dataset): |
| """ |
| Returns True if the specified event is present in the parsed trace, |
| False otherwise. |
| |
| :param dataset: trace event name or list of trace events |
| :type dataset: str or list(str) |
| """ |
| if dataset in self.available_events: |
| return True |
| return False |
| |
| def __computeTimeSpan(self): |
| """ |
| Compute time axis range, considering all the parsed events. |
| """ |
| ts = sys.maxint |
| te = 0 |
| |
| for events in self.available_events: |
| df = self._dfg_trace_event(events) |
| if len(df) == 0: |
| continue |
| if (df.index[0]) < ts: |
| ts = df.index[0] |
| if (df.index[-1]) > te: |
| te = df.index[-1] |
| self.time_range = te - ts |
| |
| self._log.debug('Collected events spans a %.3f [s] time interval', |
| self.time_range) |
| |
| # Build a stat on trace overutilization |
| if self.hasEvents('sched_overutilized'): |
| df = self._dfg_trace_event('sched_overutilized') |
| self.overutilized_time = df[df.overutilized == 1].len.sum() |
| self.overutilized_prc = 100. * self.overutilized_time / self.time_range |
| |
| self._log.debug('Overutilized time: %.6f [s] (%.3f%% of trace time)', |
| self.overutilized_time, self.overutilized_prc) |
| |
| def _scanTgids(self, df): |
| if not '__tgid' in df.columns: |
| return |
| df = df[['__pid', '__tgid']] |
| df = df.drop_duplicates(keep='first').set_index('__pid') |
| df.rename(columns = { '__pid': 'pid', '__tgid': 'tgid' }, |
| inplace=True) |
| self._pid_tgid = df |
| |
| def _scanTasks(self, df, name_key='comm', pid_key='pid'): |
| """ |
| Extract tasks names and PIDs from the input data frame. The data frame |
| should contain a task name column and PID column. |
| |
| :param df: data frame containing trace events from which tasks names |
| and PIDs will be extracted |
| :type df: :mod:`pandas.DataFrame` |
| |
| :param name_key: The name of the dataframe columns containing task |
| names |
| :type name_key: str |
| |
| :param pid_key: The name of the dataframe columns containing task PIDs |
| :type pid_key: str |
| """ |
| df = df[[name_key, pid_key]].drop_duplicates() |
| self._tasks_by_name = df.set_index(name_key) |
| self._tasks_by_pid = df.set_index(pid_key) |
| |
| def getTaskByName(self, name): |
| """ |
| Get the PIDs of all tasks with the specified name. |
| |
| :param name: task name |
| :type name: str |
| """ |
| if name not in self._tasks_by_name.index: |
| return [] |
| if len(self._tasks_by_name.ix[name].values) > 1: |
| return list({task[0] for task in |
| self._tasks_by_name.ix[name].values}) |
| return [self._tasks_by_name.ix[name].values[0]] |
| |
| def getTaskByPid(self, pid): |
| """ |
| Get the names of all tasks with the specified PID. |
| |
| :param name: task PID |
| :type name: int |
| """ |
| if pid not in self._tasks_by_pid.index: |
| return [] |
| if len(self._tasks_by_pid.ix[pid].values) > 1: |
| return list({task[0] for task in |
| self._tasks_by_pid.ix[pid].values}) |
| return [self._tasks_by_pid.ix[pid].values[0]] |
| |
| def getTgidFromPid(self, pid): |
| return _pid_tgid.ix[pid].values[0] |
| |
| def getTasks(self, dataframe=None, |
| task_names=None, name_key='comm', pid_key='pid'): |
| """ |
| Helper function to get PIDs of specified tasks. |
| |
| This method can take a Pandas dataset in input to be used to fiter out |
| the PIDs of all the specified tasks. If a dataset is not provided, |
| previously filtered PIDs are returned. |
| |
| If a list of task names is not provided, all tasks detected in the trace |
| will be used. The specified dataframe must provide at least two columns |
| reporting the task name and the task PID. The default values of this |
| colums could be specified using the provided parameters. |
| |
| :param dataframe: A Pandas dataframe containing at least 'name_key' and |
| 'pid_key' columns. If None, the all PIDs are returned. |
| :type dataframe: :mod:`pandas.DataFrame` |
| |
| :param task_names: The list of tasks to get the PID of (default: all |
| tasks) |
| :type task_names: list(str) |
| |
| :param name_key: The name of the dataframe columns containing task |
| names |
| :type name_key: str |
| |
| :param pid_key: The name of the dataframe columns containing task PIDs |
| :type pid_key: str |
| """ |
| if task_names is None: |
| task_names = self.tasks.keys() |
| if dataframe is None: |
| return {k: v for k, v in self.tasks.iteritems() if k in task_names} |
| df = dataframe |
| self._log.debug('Lookup dataset for tasks...') |
| for tname in task_names: |
| self._log.debug('Lookup for task [%s]...', tname) |
| results = df[df[name_key] == tname][[name_key, pid_key]] |
| if len(results) == 0: |
| self._log.error(' task %16s NOT found', tname) |
| continue |
| (name, pid) = results.head(1).values[0] |
| if name != tname: |
| self._log.error(' task %16s NOT found', tname) |
| continue |
| if tname not in self.tasks: |
| self.tasks[tname] = {} |
| pids = list(results[pid_key].unique()) |
| self.tasks[tname]['pid'] = pids |
| self._log.debug(' task %16s found, pid: %s', |
| tname, self.tasks[tname]['pid']) |
| return self.tasks |
| |
| |
| ############################################################################### |
| # DataFrame Getter Methods |
| ############################################################################### |
| |
| def df(self, event): |
| """ |
| Get a dataframe containing all occurrences of the specified trace event |
| in the parsed trace. |
| |
| :param event: Trace event name |
| :type event: str |
| """ |
| warnings.simplefilter('always', DeprecationWarning) #turn off filter |
| warnings.warn("\n\tUse of Trace::df() is deprecated and will be soon removed." |
| "\n\tUse Trace::data_frame.trace_event(event_name) instead.", |
| category=DeprecationWarning) |
| warnings.simplefilter('default', DeprecationWarning) #reset filter |
| return self._dfg_trace_event(event) |
| |
| def _dfg_trace_event(self, event): |
| """ |
| Get a dataframe containing all occurrences of the specified trace event |
| in the parsed trace. |
| |
| :param event: Trace event name |
| :type event: str |
| """ |
| if self.data_dir is None: |
| raise ValueError("trace data not (yet) loaded") |
| if self.ftrace and hasattr(self.ftrace, event): |
| return getattr(self.ftrace, event).data_frame |
| raise ValueError('Event [{}] not supported. ' |
| 'Supported events are: {}' |
| .format(event, self.available_events)) |
| |
| def _dfg_functions_stats(self, functions=None): |
| """ |
| Get a DataFrame of specified kernel functions profile data |
| |
| For each profiled function a DataFrame is returned which reports stats |
| on kernel functions execution time. The reported stats are per-CPU and |
| includes: number of times the function has been executed (hits), |
| average execution time (avg), overall execution time (time) and samples |
| variance (s_2). |
| By default returns a DataFrame of all the functions profiled. |
| |
| :param functions: the name of the function or a list of function names |
| to report |
| :type functions: str or list(str) |
| """ |
| if not hasattr(self, '_functions_stats_df'): |
| return None |
| df = self._functions_stats_df |
| if not functions: |
| return df |
| return df.loc[df.index.get_level_values(1).isin(listify(functions))] |
| |
| # cgroup_attach_task with just merged fake and real events |
| def _cgroup_attach_task(self): |
| cgroup_events = ['cgroup_attach_task', 'cgroup_attach_task_devlib'] |
| df = None |
| |
| if set(cgroup_events).isdisjoint(set(self.available_events)): |
| self._log.error('atleast one of {} is needed for cgroup_attach_task event generation'.format(cgroup_events)) |
| return None |
| |
| for cev in cgroup_events: |
| if not cev in self.available_events: |
| continue |
| cdf = self._dfg_trace_event(cev) |
| cdf = cdf[['__line', 'pid', 'controller', 'cgroup']] |
| if not isinstance(df, pd.DataFrame): |
| df = cdf |
| else: |
| df = pd.concat([cdf, df]) |
| |
| # Always drop na since this DF is used as secondary |
| df.dropna(inplace=True, how='any') |
| return df |
| |
| @memoized |
| def _dfg_cgroup_attach_task(self, controllers = ['schedtune', 'cpuset']): |
| # Since fork doesn't result in attach events, generate fake attach events |
| # The below mechanism doesn't work to propogate nested fork levels: |
| # For ex: |
| # cgroup_attach_task: pid=1166 |
| # fork: pid=1166 child_pid=2222 <-- fake attach generated |
| # fork: pid=2222 child_pid=3333 <-- fake attach not generated |
| def fork_add_cgroup(fdf, cdf, controller): |
| cdf = cdf[cdf['controller'] == controller] |
| ret_df = trappy.utils.merge_dfs(fdf, cdf, pivot='pid') |
| return ret_df |
| |
| if not 'sched_process_fork' in self.available_events: |
| self._log.error('sched_process_fork is mandatory to get proper cgroup_attach events') |
| return None |
| fdf = self._dfg_trace_event('sched_process_fork') |
| |
| forks_len = len(fdf) |
| forkdf = fdf |
| cdf = self._cgroup_attach_task() |
| for idx, c in enumerate(controllers): |
| fdf = fork_add_cgroup(fdf, cdf, c) |
| if (idx != (len(controllers) - 1)): |
| fdf = pd.concat([fdf, forkdf]).sort_values(by='__line') |
| |
| fdf = fdf[['__line', 'child_pid', 'controller', 'cgroup']] |
| fdf.rename(columns = { 'child_pid': 'pid' }, inplace=True) |
| |
| # Always drop na since this DF is used as secondary |
| fdf.dropna(inplace=True, how='any') |
| |
| new_forks_len = len(fdf) / len(controllers) |
| |
| fdf = pd.concat([fdf, cdf]).sort_values(by='__line') |
| |
| if new_forks_len < forks_len: |
| dropped = forks_len - new_forks_len |
| self._log.info("Couldn't attach all forks cgroup with-attach events ({} dropped)".format(dropped)) |
| return fdf |
| |
| @memoized |
| def _dfg_sched_switch_cgroup(self, controllers = ['schedtune', 'cpuset']): |
| def sched_switch_add_cgroup(sdf, cdf, controller, direction): |
| cdf = cdf[cdf['controller'] == controller] |
| |
| ret_df = sdf.rename(columns = { direction + '_pid': 'pid' }) |
| ret_df = trappy.utils.merge_dfs(ret_df, cdf, pivot='pid') |
| ret_df.rename(columns = { 'pid': direction + '_pid' }, inplace=True) |
| |
| ret_df.drop('controller', axis=1, inplace=True) |
| ret_df.rename(columns = { 'cgroup': direction + '_' + controller }, inplace=True) |
| return ret_df |
| |
| if not 'sched_switch' in self.available_events: |
| self._log.error('sched_switch is mandatory to generate sched_switch_cgroup event') |
| return None |
| sdf = self._dfg_trace_event('sched_switch') |
| cdf = self._dfg_cgroup_attach_task() |
| |
| for c in controllers: |
| sdf = sched_switch_add_cgroup(sdf, cdf, c, 'next') |
| sdf = sched_switch_add_cgroup(sdf, cdf, c, 'prev') |
| |
| # Augment with TGID information |
| sdf = sdf.join(self._pid_tgid, on='next_pid').rename(columns = {'tgid': 'next_tgid'}) |
| sdf = sdf.join(self._pid_tgid, on='prev_pid').rename(columns = {'tgid': 'prev_tgid'}) |
| |
| df = self._tasks_by_pid.rename(columns = { 'next_comm': 'comm' }) |
| sdf = sdf.join(df, on='next_tgid').rename(columns = {'comm': 'next_tgid_comm'}) |
| sdf = sdf.join(df, on='prev_tgid').rename(columns = {'comm': 'prev_tgid_comm'}) |
| return sdf |
| |
| ############################################################################### |
| # Trace Events Sanitize Methods |
| ############################################################################### |
| |
| def _sanitize_SchedCpuCapacity(self): |
| """ |
| Add more columns to cpu_capacity data frame if the energy model is |
| available. |
| """ |
| if not self.hasEvents('cpu_capacity') \ |
| or 'nrg_model' not in self.platform: |
| return |
| |
| df = self._dfg_trace_event('cpu_capacity') |
| |
| # Add column with LITTLE and big CPUs max capacities |
| nrg_model = self.platform['nrg_model'] |
| max_lcap = nrg_model['little']['cpu']['cap_max'] |
| max_bcap = nrg_model['big']['cpu']['cap_max'] |
| df['max_capacity'] = np.select( |
| [df.cpu.isin(self.platform['clusters']['little'])], |
| [max_lcap], max_bcap) |
| # Add LITTLE and big CPUs "tipping point" threshold |
| tip_lcap = 0.8 * max_lcap |
| tip_bcap = 0.8 * max_bcap |
| df['tip_capacity'] = np.select( |
| [df.cpu.isin(self.platform['clusters']['little'])], |
| [tip_lcap], tip_bcap) |
| |
| def _sanitize_SchedLoadAvgCpu(self): |
| """ |
| If necessary, rename certain signal names from v5.0 to v5.1 format. |
| """ |
| if not self.hasEvents('sched_load_avg_cpu'): |
| return |
| df = self._dfg_trace_event('sched_load_avg_cpu') |
| if 'utilization' in df: |
| df.rename(columns={'utilization': 'util_avg'}, inplace=True) |
| df.rename(columns={'load': 'load_avg'}, inplace=True) |
| |
| def _sanitize_SchedLoadAvgTask(self): |
| """ |
| If necessary, rename certain signal names from v5.0 to v5.1 format. |
| """ |
| if not self.hasEvents('sched_load_avg_task'): |
| return |
| df = self._dfg_trace_event('sched_load_avg_task') |
| if 'utilization' in df: |
| df.rename(columns={'utilization': 'util_avg'}, inplace=True) |
| df.rename(columns={'load': 'load_avg'}, inplace=True) |
| df.rename(columns={'avg_period': 'period_contrib'}, inplace=True) |
| df.rename(columns={'runnable_avg_sum': 'load_sum'}, inplace=True) |
| df.rename(columns={'running_avg_sum': 'util_sum'}, inplace=True) |
| df['cluster'] = np.select( |
| [df.cpu.isin(self.platform['clusters']['little'])], |
| ['LITTLE'], 'big') |
| # Add a column which represents the max capacity of the smallest |
| # clustre which can accomodate the task utilization |
| little_cap = self.platform['nrg_model']['little']['cpu']['cap_max'] |
| big_cap = self.platform['nrg_model']['big']['cpu']['cap_max'] |
| df['min_cluster_cap'] = df.util_avg.map( |
| lambda util_avg: big_cap if util_avg > little_cap else little_cap |
| ) |
| |
| def _sanitize_SchedBoostCpu(self): |
| """ |
| Add a boosted utilization signal as the sum of utilization and margin. |
| |
| Also, if necessary, rename certain signal names from v5.0 to v5.1 |
| format. |
| """ |
| if not self.hasEvents('sched_boost_cpu'): |
| return |
| df = self._dfg_trace_event('sched_boost_cpu') |
| if 'usage' in df: |
| df.rename(columns={'usage': 'util'}, inplace=True) |
| df['boosted_util'] = df['util'] + df['margin'] |
| |
| def _sanitize_SchedBoostTask(self): |
| """ |
| Add a boosted utilization signal as the sum of utilization and margin. |
| |
| Also, if necessary, rename certain signal names from v5.0 to v5.1 |
| format. |
| """ |
| if not self.hasEvents('sched_boost_task'): |
| return |
| df = self._dfg_trace_event('sched_boost_task') |
| if 'utilization' in df: |
| # Convert signals name from to v5.1 format |
| df.rename(columns={'utilization': 'util'}, inplace=True) |
| df['boosted_util'] = df['util'] + df['margin'] |
| |
| def _sanitize_SchedEnergyDiff(self): |
| """ |
| If a energy model is provided, some signals are added to the |
| sched_energy_diff trace event data frame. |
| |
| Also convert between existing field name formats for sched_energy_diff |
| """ |
| if not self.hasEvents('sched_energy_diff') \ |
| or 'nrg_model' not in self.platform: |
| return |
| nrg_model = self.platform['nrg_model'] |
| em_lcluster = nrg_model['little']['cluster'] |
| em_bcluster = nrg_model['big']['cluster'] |
| em_lcpu = nrg_model['little']['cpu'] |
| em_bcpu = nrg_model['big']['cpu'] |
| lcpus = len(self.platform['clusters']['little']) |
| bcpus = len(self.platform['clusters']['big']) |
| SCHED_LOAD_SCALE = 1024 |
| |
| power_max = em_lcpu['nrg_max'] * lcpus + em_bcpu['nrg_max'] * bcpus + \ |
| em_lcluster['nrg_max'] + em_bcluster['nrg_max'] |
| self._log.debug( |
| "Maximum estimated system energy: {0:d}".format(power_max)) |
| |
| df = self._dfg_trace_event('sched_energy_diff') |
| |
| translations = {'nrg_d' : 'nrg_diff', |
| 'utl_d' : 'usage_delta', |
| 'payoff' : 'nrg_payoff' |
| } |
| df.rename(columns=translations, inplace=True) |
| |
| df['nrg_diff_pct'] = SCHED_LOAD_SCALE * df.nrg_diff / power_max |
| |
| # Tag columns by usage_delta |
| ccol = df.usage_delta |
| df['usage_delta_group'] = np.select( |
| [ccol < 150, ccol < 400, ccol < 600], |
| ['< 150', '< 400', '< 600'], '>= 600') |
| |
| # Tag columns by nrg_payoff |
| ccol = df.nrg_payoff |
| df['nrg_payoff_group'] = np.select( |
| [ccol > 2e9, ccol > 0, ccol > -2e9], |
| ['Optimal Accept', 'SchedTune Accept', 'SchedTune Reject'], |
| 'Suboptimal Reject') |
| |
| def _sanitize_SchedOverutilized(self): |
| """ Add a column with overutilized status duration. """ |
| if not self.hasEvents('sched_overutilized'): |
| return |
| df = self._dfg_trace_event('sched_overutilized') |
| df['start'] = df.index |
| df['len'] = (df.start - df.start.shift()).fillna(0).shift(-1) |
| df.drop('start', axis=1, inplace=True) |
| |
| # Sanitize cgroup information helper |
| def _helper_sanitize_CgroupAttachTask(self, df, allowed_cgroups, controller_id_name): |
| # Drop rows that aren't in the root-id -> name map |
| df = df[df['dst_root'].isin(controller_id_name.keys())] |
| |
| def get_cgroup_name(path, valid_names): |
| name = os.path.basename(path) |
| name = 'root' if not name in valid_names else name |
| return name |
| |
| def get_cgroup_names(rows): |
| ret = [] |
| for r in rows.iterrows(): |
| ret.append(get_cgroup_name(r[1]['dst_path'], allowed_cgroups)) |
| return ret |
| |
| def get_controller_names(rows): |
| ret = [] |
| for r in rows.iterrows(): |
| ret.append(controller_id_name[r[1]['dst_root']]) |
| return ret |
| |
| # Sanitize cgroup names |
| # cgroup column isn't in mainline, add it in |
| # its already added for some out of tree kernels so check first |
| if not 'cgroup' in df.columns: |
| if not 'dst_path' in df.columns: |
| raise RuntimeError('Cant santize cgroup DF, need dst_path') |
| df = df.assign(cgroup = get_cgroup_names) |
| |
| # Sanitize controller names |
| if not 'controller' in df.columns: |
| if not 'dst_root' in df.columns: |
| raise RuntimeError('Cant santize cgroup DF, need dst_path') |
| df = df.assign(controller = get_controller_names) |
| |
| return df |
| |
| def _sanitize_CgroupAttachTask(self): |
| def sanitize_cgroup_event(name): |
| if not name in self.available_events: |
| return |
| |
| df = self._dfg_trace_event(name) |
| |
| if len(df.groupby(level=0).filter(lambda x: len(x) > 1)) > 0: |
| self._log.warning('Timstamp Collisions seen in {} event!'.format(name)) |
| |
| df = self._helper_sanitize_CgroupAttachTask(df, self.cgroup_info['cgroups'], |
| self.cgroup_info['controller_ids']) |
| getattr(self.ftrace, name).data_frame = df |
| sanitize_cgroup_event('cgroup_attach_task') |
| sanitize_cgroup_event('cgroup_attach_task_devlib') |
| |
| def _chunker(self, seq, size): |
| """ |
| Given a data frame or a series, generate a sequence of chunks of the |
| given size. |
| |
| :param seq: data to be split into chunks |
| :type seq: :mod:`pandas.Series` or :mod:`pandas.DataFrame` |
| |
| :param size: size of each chunk |
| :type size: int |
| """ |
| return (seq.iloc[pos:pos + size] for pos in range(0, len(seq), size)) |
| |
| def _sanitize_CpuFrequency(self): |
| """ |
| Verify that all platform reported clusters are frequency coherent (i.e. |
| frequency scaling is performed at a cluster level). |
| """ |
| if not self.hasEvents('cpu_frequency_devlib'): |
| return |
| |
| devlib_freq = self._dfg_trace_event('cpu_frequency_devlib') |
| devlib_freq.rename(columns={'cpu_id':'cpu'}, inplace=True) |
| devlib_freq.rename(columns={'state':'frequency'}, inplace=True) |
| |
| df = self._dfg_trace_event('cpu_frequency') |
| clusters = self.platform['clusters'] |
| |
| # devlib always introduces fake cpu_frequency events, in case the |
| # OS has not generated cpu_frequency envets there are the only |
| # frequency events to report |
| if len(df) == 0: |
| # Register devlib injected events as 'cpu_frequency' events |
| setattr(self.ftrace.cpu_frequency, 'data_frame', devlib_freq) |
| df = devlib_freq |
| self.available_events.append('cpu_frequency') |
| |
| # make sure fake cpu_frequency events are never interleaved with |
| # OS generated events |
| else: |
| if len(devlib_freq) > 0: |
| |
| # Frequencies injection is done in a per-cluster based. |
| # This is based on the assumption that clusters are |
| # frequency choerent. |
| # For each cluster we inject devlib events only if |
| # these events does not overlaps with os-generated ones. |
| |
| # Inject "initial" devlib frequencies |
| os_df = df |
| dl_df = devlib_freq.iloc[:self.platform['cpus_count']] |
| for _,c in self.platform['clusters'].iteritems(): |
| dl_freqs = dl_df[dl_df.cpu.isin(c)] |
| os_freqs = os_df[os_df.cpu.isin(c)] |
| self._log.debug("First freqs for %s:\n%s", c, dl_freqs) |
| # All devlib events "before" os-generated events |
| self._log.debug("Min os freq @: %s", os_freqs.index.min()) |
| if os_freqs.empty or \ |
| os_freqs.index.min() > dl_freqs.index.max(): |
| self._log.debug("Insert devlib freqs for %s", c) |
| df = pd.concat([dl_freqs, df]) |
| |
| # Inject "final" devlib frequencies |
| os_df = df |
| dl_df = devlib_freq.iloc[self.platform['cpus_count']:] |
| for _,c in self.platform['clusters'].iteritems(): |
| dl_freqs = dl_df[dl_df.cpu.isin(c)] |
| os_freqs = os_df[os_df.cpu.isin(c)] |
| self._log.debug("Last freqs for %s:\n%s", c, dl_freqs) |
| # All devlib events "after" os-generated events |
| self._log.debug("Max os freq @: %s", os_freqs.index.max()) |
| if os_freqs.empty or \ |
| os_freqs.index.max() < dl_freqs.index.min(): |
| self._log.debug("Append devlib freqs for %s", c) |
| df = pd.concat([df, dl_freqs]) |
| |
| df.sort_index(inplace=True) |
| |
| setattr(self.ftrace.cpu_frequency, 'data_frame', df) |
| |
| # Frequency Coherency Check |
| for _, cpus in clusters.iteritems(): |
| cluster_df = df[df.cpu.isin(cpus)] |
| for chunk in self._chunker(cluster_df, len(cpus)): |
| f = chunk.iloc[0].frequency |
| if any(chunk.frequency != f): |
| self._log.warning('Cluster Frequency is not coherent! ' |
| 'Failure in [cpu_frequency] events at:') |
| self._log.warning(chunk) |
| self.freq_coherency = False |
| return |
| self._log.info('Platform clusters verified to be Frequency coherent') |
| |
| ############################################################################### |
| # Utility Methods |
| ############################################################################### |
| |
| def integrate_square_wave(self, sq_wave): |
| """ |
| Compute the integral of a square wave time series. |
| |
| :param sq_wave: square wave assuming only 1.0 and 0.0 values |
| :type sq_wave: :mod:`pandas.Series` |
| """ |
| sq_wave.iloc[-1] = 0.0 |
| # Compact signal to obtain only 1-0-1-0 sequences |
| comp_sig = sq_wave.loc[sq_wave.shift() != sq_wave] |
| # First value for computing the difference must be a 1 |
| if comp_sig.iloc[0] == 0.0: |
| return sum(comp_sig.iloc[2::2].index - comp_sig.iloc[1:-1:2].index) |
| else: |
| return sum(comp_sig.iloc[1::2].index - comp_sig.iloc[:-1:2].index) |
| |
| def _loadFunctionsStats(self, path='trace.stats'): |
| """ |
| Read functions profiling file and build a data frame containing all |
| relevant data. |
| |
| :param path: path to the functions profiling trace file |
| :type path: str |
| """ |
| if os.path.isdir(path): |
| path = os.path.join(path, 'trace.stats') |
| if path.endswith('dat') or path.endswith('html'): |
| pre, ext = os.path.splitext(path) |
| path = pre + '.stats' |
| if not os.path.isfile(path): |
| return False |
| |
| # Opening functions profiling JSON data file |
| self._log.debug('Loading functions profiling data from [%s]...', path) |
| with open(os.path.join(path), 'r') as fh: |
| trace_stats = json.load(fh) |
| |
| # Build DataFrame of function stats |
| frames = {} |
| for cpu, data in trace_stats.iteritems(): |
| frames[int(cpu)] = pd.DataFrame.from_dict(data, orient='index') |
| |
| # Build and keep track of the DataFrame |
| self._functions_stats_df = pd.concat(frames.values(), |
| keys=frames.keys()) |
| |
| return len(self._functions_stats_df) > 0 |
| |
| @memoized |
| def getCPUActiveSignal(self, cpu): |
| """ |
| Build a square wave representing the active (i.e. non-idle) CPU time, |
| i.e.: |
| |
| cpu_active[t] == 1 if the CPU is reported to be non-idle by cpuidle at |
| time t |
| cpu_active[t] == 0 otherwise |
| |
| :param cpu: CPU ID |
| :type cpu: int |
| |
| :returns: A :mod:`pandas.Series` or ``None`` if the trace contains no |
| "cpu_idle" events |
| """ |
| if not self.hasEvents('cpu_idle'): |
| self._log.warning('Events [cpu_idle] not found, ' |
| 'cannot compute CPU active signal!') |
| return None |
| |
| idle_df = self._dfg_trace_event('cpu_idle') |
| cpu_df = idle_df[idle_df.cpu_id == cpu] |
| |
| cpu_active = cpu_df.state.apply( |
| lambda s: 1 if s == NON_IDLE_STATE else 0 |
| ) |
| |
| start_time = 0.0 |
| if not self.ftrace.normalized_time: |
| start_time = self.ftrace.basetime |
| |
| if cpu_active.empty: |
| cpu_active = pd.Series([0], index=[start_time]) |
| elif cpu_active.index[0] != start_time: |
| entry_0 = pd.Series(cpu_active.iloc[0] ^ 1, index=[start_time]) |
| cpu_active = pd.concat([entry_0, cpu_active]) |
| |
| # Fix sequences of wakeup/sleep events reported with the same index |
| return handle_duplicate_index(cpu_active) |
| |
| |
| @memoized |
| def getClusterActiveSignal(self, cluster): |
| """ |
| Build a square wave representing the active (i.e. non-idle) cluster |
| time, i.e.: |
| |
| cluster_active[t] == 1 if at least one CPU is reported to be non-idle |
| by CPUFreq at time t |
| cluster_active[t] == 0 otherwise |
| |
| :param cluster: list of CPU IDs belonging to a cluster |
| :type cluster: list(int) |
| |
| :returns: A :mod:`pandas.Series` or ``None`` if the trace contains no |
| "cpu_idle" events |
| """ |
| if not self.hasEvents('cpu_idle'): |
| self._log.warning('Events [cpu_idle] not found, ' |
| 'cannot compute cluster active signal!') |
| return None |
| |
| active = self.getCPUActiveSignal(cluster[0]).to_frame(name=cluster[0]) |
| for cpu in cluster[1:]: |
| active = active.join( |
| self.getCPUActiveSignal(cpu).to_frame(name=cpu), |
| how='outer' |
| ) |
| |
| active.fillna(method='ffill', inplace=True) |
| |
| # Cluster active is the OR between the actives on each CPU |
| # belonging to that specific cluster |
| cluster_active = reduce( |
| operator.or_, |
| [cpu_active.astype(int) for _, cpu_active in |
| active.iteritems()] |
| ) |
| |
| return cluster_active |
| |
| |
| class TraceData: |
| """ A DataFrame collector exposed to Trace's clients """ |
| pass |
| |
| # vim :set tabstop=4 shiftwidth=4 expandtab |
