text: # 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 shutil
  import uuid
  import sys
  import re
  import gc
  import math
  import abc
  import copy
  import io
  import os
  import os.path
  import json
  import warnings
  import inspect
  import shlex
  import contextlib
  import tempfile
  from functools import lru_cache, reduce, wraps
  from collections.abc import Iterable, Set, Mapping, Sequence
  from collections import namedtuple, OrderedDict
  from operator import itemgetter
  from numbers import Number, Integral, Real
  import multiprocessing
  import textwrap
  import subprocess
  import itertools
  
  import pandas as pd
  import numpy as np
  import pandas as pd
  import pyarrow.lib
  
  import devlib
  from devlib.target import KernelVersion
  
  import lisa.utils
  from lisa.utils import Loggable, HideExekallID, memoized, lru_memoized, deduplicate, take, deprecate, nullcontext, measure_time, checksum, newtype, groupby, take
  from lisa.conf import SimpleMultiSrcConf, KeyDesc, TopLevelKeyDesc, Configurable
  from lisa.generic import TypedList
  from lisa.datautils import df_split_signals, df_window, df_window_signals, SignalDesc, df_add_delta, df_combine_duplicates, series_convert, df_deduplicate
  from lisa.version import VERSION_TOKEN
  from lisa.typeclass import FromString, IntListFromStringInstance
  
  
  class TaskID(namedtuple('TaskID', ('pid', 'comm'))):
      """
      Unique identifier of a logical task in a :class:`Trace`.
  
      :param pid: PID of the task. ``None`` indicates the PID is not important.
      :type pid: int
  
      :param comm: Name of the task. ``None`` indicates the name is not important.
          This is useful to describe tasks like PID0, which can have multiple
          names associated.
      :type comm: str
      """
  
      # Prevent creation of a __dict__. This allows a more compact representation
      __slots__ = []
  
      def __init__(self, *args, **kwargs):
          # This happens when the number of saved PID/comms entries in the trace
          # is too low
          if self.comm == '<...>':
              raise ValueError('Invalid comm name "<...>", please increase saved_cmdlines_nr value on FtraceCollector')
  
      def __str__(self):
          if self.pid is not None and self.comm is not None:
              out = '{}:{}'.format(self.pid, self.comm)
          else:
              out = str(self.comm if self.comm is not None else self.pid)
  
          return '[{}]'.format(out)
  
      _STR_PARSE_REGEX = re.compile(r'\[?([0-9]+):([a-zA-Z0-9_-]+)\]?')
  
  
  class TaskIDFromStringInstance(FromString, types=TaskID):
      """
      Instance of :class:`lisa.typeclass.FromString` for :class:`TaskID` type.
      """
      @classmethod
      def from_str(cls, string):
          try:
              pid = int(string)
              comm = None
          except ValueError:
              match = cls._STR_PARSE_REGEX.match(string)
              if match:
                  pid = int(match.group(1))
                  comm = match.group(2)
              else:
                  pid = None
                  comm = string
  
          return cls(pid=pid, comm=comm)
  
      @classmethod
      def get_format_description(cls, short):
          if short:
              return 'task ID'
          else:
              return textwrap.dedent("""
              Can be any of:
                 * a PID
                 * a task name
                 * a PID (first) and a name (second): pid:name
              """).strip()
  
  
  class TaskIDListFromStringInstance(FromString, types=TypedList[TaskID]):
      """
      Instance of :class:`lisa.typeclass.FromString` for lists :class:`TaskID` type.
      """
      @classmethod
      def from_str(cls, string):
          """
          The format is a comma-separated list of :class:`TaskID`.
          """
          from_str = FromString(TaskID).from_str
          return [
              from_str(string.strip())
              for string in string.split(',')
          ]
  
      @classmethod
      def get_format_description(cls, short):
          return 'comma-separated TaskIDs'
  
  
  CPU = newtype(int, 'CPU', doc='Alias to ``int`` used for CPU IDs')
  
  
  class CPUListFromStringInstance(FromString, types=TypedList[CPU]):
      # Use the same implementation as for TypedList[int]
      from_str = IntListFromStringInstance.from_str
  
      @classmethod
      def get_format_description(cls, short):
          return FromString(TypedList[int]).get_format_description(short=short)
  
  
  class MissingMetadataError(KeyError):
      """
      Raised when a given metadata is not available.
      """
      def __init__(self, metadata):
          self.metadata = metadata
  
      def __str__(self):
          return 'Missing metadata: {}'.format(self.metadata)
  
  
  class TraceParserBase(abc.ABC, Loggable):
      """
      Abstract Base Class for trace parsers.
  
      :param events: Iterable of events to parse. An empty iterable can be
          passed, in which case some metadata may still be available.
      :param events: collections.abc.Iterable(str)
  
      :param needed_metadata: Set of metadata name to gather in the parser.
      :type needed_metadata: collections.abc.Iterable(str)
      """
  
      def __init__(self, events, needed_metadata):
          self._needed_metadata = set(needed_metadata or [])
  
      def get_metadata(self, key):
          """
          Return the metadata value.
  
          :param key: Name of the metadata. Can be one of:
  
              * ``time-range``: tuple ``(start, end)`` of the timestamps in the
                trace. This must be the first timestamp to appear in the trace,
                regardless of what events is being parsed. Otherwise, it would be
                impossible to use the time range of a parser in the mother
                :class:`TraceBase` when requesting specific events.
  
              * ``symbols-address``: Dictionnary of address (int) to symbol names
                in the kernel (str) that was used to create the trace. This
                allows resolving the fields of events that recorded addresses
                rather than function names.
  
              * ``cpus-count``: Number of CPUs on the system the trace was
                collected on.
  
              * ``available-events``: List of all available events stored in the
                trace. The list must be exhaustive, not limited to the events
                that were requested.
          :type key: str
  
          :raises: :exc:`MissingMetadataError` if the metadata is not available
              on that parser.
  
          .. note:: A given metadata can only be expected to be available if
              asked for in the constructor, but bear in mind that there is no
              promise on the availability of any except for the following that
              *must* be provided if asked for:
  
                  * ``time-range``
  
              Metadata may still be made available if not asked for, but only if
              it's a very cheap byproduct of parsing that incurs no extra cost.
          """
          raise MissingMetadataError(key)
  
      @abc.abstractmethod
      def parse_event(self, event):
          """
          Parse the given event from the trace and return a
          :class:`pandas.DataFrame` with the following columns:
  
              * ``Time`` index: floating point absolute timestamp in seconds. The
                index *must not* have any duplicated values.
              * One column per event field, with the appropriate dtype.
              * Columns prefixed with ``__``: Header of each event, usually containing the following fields:
  
                  * ``__cpu``: CPU number the event was emitted from
                  * ``__pid``: PID of the current process scheduled at the time the event was emitted
                  * ``__comm``: Task command name going with ``__pid`` at the point the event was emitted
  
          :param event: name of the event to parse
          :type event: str
  
          :raises MissingTraceEventError: If the event cannot be parsed.
  
          .. note:: The caller is free to modify the index of the data, and it
              must not affect other dataframes.
          """
          pass
  
      def parse_events(self, events, best_effort=False, **kwargs):
          """
          Same as :meth:`parse_event` but taking a list of events as input, and
          returning a mapping of event names to :class:`pandas.DataFrame` for
          each.
  
          :param events: Event names to parse.
          :type events: list(str)
  
          :param best_effort: If ``True``, do not raise
              :exc:`MissingTraceEventError`, silently skip the event.
              Must default to ``False``.
          :type best_effort: bool
  
          :param kwargs: See :meth:`parse_event`
          """
  
          def parse(event):
              try:
                  return self.parse_event(event)
              except MissingTraceEventError:
                  if best_effort:
                      return None
                  else:
                      raise
          return {
              event: df
              for event, df in (
                  (event, parse(event))
                  for event in events
              )
              if df is not None
          }
  
  
  class EventParserBase:
      """
      Base class for trace event parser.
  
      Required attributes or properties:
  
          * ``event``: name of the event
          * ``regex``: full regex to parse a line of the event
          * ``fields``: mapping of field names to :mod:`pandas` dtype to use for
            the :class:`pandas.DataFrame` column.
      """
  
      PARSER_REGEX_TERMINALS = dict(
          # To be used with the re.ASCII regex flag
          identifier=r'\w+',
          integer=r'\d+',
          floating=r'\d+\.\d+',
          blank=r' +',
      )
      """
      Snippets of regex to be used in building more complex regexes in textual trace parsers.
  
      .. note:: Meant to be used with the :data:`re.ASCII` flags.
      """
  
      def __init__(self, event, fields):
          self.event = event
          self.fields = fields
  
  
  class TxtEventParser(EventParserBase):
      """
      Trace event parser for raw output of ``trace-cmd report -R trace.dat``.
  
      :param event: name of the event
      :type event: str
  
      :param fields: mapping of field name to :class:`pandas.DataFrame` column
          dtype to use for each.
      :type fields: dict(str, str)
  
      :param positional_field: Name of the positional field. If ``None``, no
          positional field will be parsed.
      :type positional_field: str or None
  
      :param greedy_field: Name of a greedy field that will consume the
          remainder of the line, no matter what the content is. This allows
          parsing events with a field containing a string formatted itself as
          an event
      :type greedy_field: str or None
  
      :param raw: If ``True``, `trace-cmd report` will be used in raw mode. This
          usually ensures compliance with the format, but may sometimes be a problem.
          For exampe ``const char*`` are displayed as an hex pointer in raw mode,
          which is not helpful.
      :type raw: bool
  
      Parses events with the following format::
  
            <idle>-0     [001]    76.214046: sched_wakeup: something here: comm=watchdog/1 pid=15 prio=0 success=1 target_cpu=1
            \____________________________________________/ \____________/  \__________________________________________________/
                               header                        positional                          fields
  
      """
  
      def __init__(self, event, fields, positional_field=None, greedy_field=None, raw=True):
          super().__init__(
              event=event,
              fields=fields,
          )
          regex = self._get_regex(event, fields, positional_field, greedy_field)
          self.regex = re.compile(regex, flags=re.ASCII)
          self.raw = raw
  
      @property
      def bytes_regex(self):
          """
          Same as ``regex`` but acting on :class:`bytes` instead of :class:`str`.
          """
          regex = self.regex
          return re.compile(
              self.regex.pattern.encode('ascii'),
              flags=regex.flags,
          )
  
      @classmethod
      def _get_fields_regex(cls, event, fields, positional_field, greedy_field):
          """
          Returns the regex to parse the fields part of the event line.
  
          :param event: name of the event
          :type event: str
  
          :param fields: Mapping of field names to dataframe column dtype
          :type fields: dict(str, str)
  
          :param positional_field: Name to give to the positional field column,
              or ``None`` if it should not be parsed.
          :type positional_field: str or None
  
          :param greedy_field: Name of a greedy field that will consume the
              remainder of the line, no matter what the content is. This allows
              parsing events with a field containing a string formatted itself as
              an event
          :type greedy_field: str or None
          """
          fields = fields.keys() - {positional_field}
  
          if fields:
              def combine(fields):
                  return r'(?:{})+'.format(
                      '|'.join(fields)
                  )
  
              def make_regex(field):
                  if field == greedy_field:
                      return r'{field}=(?P<{field}>.*)'.format(
                          field=re.escape(field),
                          **cls.PARSER_REGEX_TERMINALS
                      )
                  else:
                      # The non-capturing group with positive lookahead is
                      # necessary to be able to correctly collect spaces in the
                      # values of fields
                      return r'{field}=(?P<{field}>.+?)(?:{blank}(?={identifier}=)|$)'.format(
                          field=re.escape(field),
                          **cls.PARSER_REGEX_TERMINALS
                      )
  
              fields_regexes = list(map(make_regex, fields))
  
              # Catch-all field that will consume any unknown field, allowing for
              # partial parsing (both for performance/memory consumption and
              # forward compatibility)
              fields_regexes.append(r'{identifier}=.*?(?=(?:{other_fields})=)'.format(
                  other_fields='|'.join(fields),
                  **cls.PARSER_REGEX_TERMINALS
              ))
  
              fields = combine(fields_regexes)
          else:
              fields = ''
  
          if positional_field:
              # If there are more fields to match, use the first ":" or spaces as
              # separator, otherwise just consume everything
              if fields:
                  fields =  r' *:? *{fields}'.format(fields=fields)
  
              fields = r'(?P<{pos}>.*?){fields}$'.format(pos=positional_field, fields=fields, **cls.PARSER_REGEX_TERMINALS)
  
          return fields
  
      @classmethod
      def _get_header_regex(cls, event):
          """
          Return the regex for the header of the event.
  
          :param event: Name of the event
          :type event: str
          """
          blank = cls.PARSER_REGEX_TERMINALS['blank']
          regex_map = dict(
              __comm=r'\S+',
              __pid=cls.PARSER_REGEX_TERMINALS['integer'],
              __cpu=cls.PARSER_REGEX_TERMINALS['integer'],
              __time=cls.PARSER_REGEX_TERMINALS['floating'],
              __event=re.escape(event),
          )
  
          compos = {
              field: r'(?P<{field}>{regex})'.format(field=field, regex=regex)
              for field, regex in regex_map.items()
          }
  
          # We don't need to capture these ones as they have already been parsed
          # in the skeleton dataframe, and fixed up for __time
          compos.update(
              (field, regex)
              for field, regex in regex_map.items()
              if field in ('__time', '__event')
          )
  
          regex = r'^{blank}{__comm}-{__pid}{blank}\[{__cpu}\]{blank}{__time}:{blank}{__event}:'.format(**compos, blank=blank)
          return regex
  
      def _get_regex(self, event, fields, positional_field, greedy_field):
          """
          Return the full regex to parse the event line.
  
          This includes both the header and the fields part.
          """
          fields = self._get_fields_regex(event, fields, positional_field, greedy_field)
          header = self._get_header_regex(event)
          return r'{header} *{fields}'.format(header=header, fields=fields, **self.PARSER_REGEX_TERMINALS)
  
  
  class PrintTxtEventParser(TxtEventParser):
      """
      Event parser for the folling events, displayed in non-raw format by
      ``trace-cmd``:
  
          * ``print``
          * ``bprint``
          * ``bputs``
  
      .. note:: ``bputs`` and ``print`` could be parsed in raw format, but that
          would make them harder to parse (function resolution needed), and
          ``bprint`` is just impossible to parse in raw format, since the data to
          interpolate the format string with are not displayed by ``trace-cmd``.
      """
      def __init__(self, event, func_field, content_field):
          fields = {
              func_field: 'string',
              # Use bytes so that we can easily parse meta events out of it
              # without conversion
              content_field: 'bytes',
          }
          self._func_field = func_field
          self._content_field = content_field
          super().__init__(event=event, fields=fields, raw=False)
  
      def _get_fields_regex(self, event, fields, positional_field, greedy_field):
          return r'(?P<{func}>.*?): *(?P<{content}>.*)'.format(
              func=self._func_field,
              content=self._content_field,
          )
  
  
  class CustomFieldsTxtEventParser(TxtEventParser):
      """
      Subclass of :class:`TxtEventParser` to be used for funky formats.
  
      When the format of the textual event does not respect at all the raw
      ``trace-cmd`` format, and if raw format cannot be used (e.g. because of
      ``const char*`` fields), this class provides a way out. For example, this
      event can be parsed with this class, but would be impossible to be parse
      using :class:`TxtEventParser`::
  
          # non-raw format lacks a field delimiter for the "reason"
          kworker/u16:6-262   [003]   177.417147: ipi_raise:            target_mask=00000000,00000020 (Function call interrupts)
          # raw format, even less usable because of the const char* pointer not being resolved
          kworker/u16:6-262   [003]   177.417147: ipi_raise:             target_cpus=ARRAY[20, 00, 00, 00, 00, 00, 00, 00] reason=0xffffff8c0774fe6b
  
      .. note:: Use :class:`TxtEventParser` if possible, since it provides a more
          optimized fields regex than what you are likely to come up with, and
          can deal with missing fields.
  
      :param event: Name of the event.
      :type event: str
  
      :param fields_regex: Regex to parse the fields part of the event occurence.
          Regex groups are used to delimit fields, e.g.
          ``r"field1=(?P<field1>[0-9]+)"`` would recognize ``"field1=42"`` as a
          ``field1`` column.
      :type fields_regex: str
  
      :param fields: Mapping of field names (group names in the regex) to dtype
          to use in the :class:`pandas.DataFrame`. This is passed to
          :func:`lisa.datautils.series_convert` so the accepted values are a bit
          wider than :mod:`pandas` dtypes.
      :type fields: dict(str, object)
  
      :param raw: If ``True``, the event will be parsed as raw by ``trace-cmd``.
          If you have ``const char*`` fields, this must be ``False`` in order to
          get the string instead of the pointer.
      :type raw: bool
      """
      def __init__(self, event, fields_regex, fields, raw):
          self._fields_regex = fields_regex
          super().__init__(event=event, fields=fields, raw=raw)
  
      def _get_fields_regex(self, event, fields, positional_field, greedy_field):
          return self._fields_regex
  
  
  class TxtTraceParserBase(TraceParserBase):
      """
      Text trace parser base class.
  
      :param lines: Iterable of text lines as :class:`bytes`.
      :type lines: collections.abc.Iterable(bytes)
  
      :param events: List of events that will be available using
          :meth:`parse_event`. If not provided, all events will be considered.
          .. note:: Restricting the set of events can speed up some operations.
      :type events: list(str)
  
      :param event_parsers: Pre-built event parsers. Missing event parsers will
          be inferred from the fields parsed in the trace, which is costly and
          can lead to using larger dtypes than necessary (e.g. ``int64`` rather
          than ``uint16``).
  
          .. seealso:: :class:`TxtEventParser`
      :type event_parsers: list(EventParserBase)
  
      :param default_event_parser_cls: Class used to build event parsers inferred from the trace.
      :type default_event_parser_cls: type
  
      :param pre_filtered_lines: Set to ``True`` if the lines provided were
          already pre-filtered externally, meaning that it will not accurately
          reflect all the events that are available in that trace.
      :type pre_filtered_lines: bool
  
      :param pre_filled_metadata: Metadata pre-filled by the caller of the
          constructor.
      :type pre_filled_metadata: dict(str, object) or None
      """
  
      HEADER_FIELDS = {
          '__comm': 'string',
          '__pid': 'uint32',
          '__cpu': 'uint32',
          '__time': 'float64',
          '__event': 'string',
      }
      """
      Pandas dtype of the header fields.
      """
  
      DTYPE_INFERENCE_ORDER = ['int64', 'uint64', 'float64', 'string']
      """
      When the dtype of a field is not provided by a user-defined parser, these
      dtypes will be tried in order to convert the column from string to
      something more appropriate.
  
      .. note:: ``uint64`` allows testing for hexadecimal formatting of numbers.
      """
  
      DEFAULT_EVENT_PARSER_CLS = None
      """
      Class used to create event parsers when inferred from the trace.
      """
  
      EVENT_DESCS = {}
      """
      Mapping of event names to parser description as a dict.
  
      Each event description can include the constructor parameters of the class
      used as :data:`DEFAULT_EVENT_PARSER_CLS`, which will be used to build event
      parsers from the descriptions.
  
      If an instance of :class:`EventParserBase` is passed instead of a dict, it
      will be used as is.
      """
  
      # Since re.Match is only importable directly since Python >= 3.7, use a
      # dummy match to get the type
      _RE_MATCH_CLS = re.match('x', 'x').__class__
  
      def __init__(self,
          lines,
          events=None,
          needed_metadata=None,
          event_parsers=None,
          default_event_parser_cls=None,
          pre_filtered_lines=False,
          pre_filled_metadata=None,
      ):
          super().__init__(events, needed_metadata=needed_metadata)
          self._pre_filtered_lines = pre_filtered_lines
          self._pre_filled_metadata = pre_filled_metadata or {}
          events = set(events or [])
  
          default_event_parser_cls, event_parsers = self._resolve_event_parsers(event_parsers, default_event_parser_cls)
  
          # Remove all the parsers that are unnecessary
          event_parsers = {
              event: parser
              for event, parser in event_parsers.items()
              if event in events
          }
  
          # If we don't need the fields in the skeleton df, avoid collecting them
          # to save memory and speed things up
          need_fields = (events != event_parsers.keys())
          skeleton_regex = self._get_skeleton_regex(need_fields)
  
          events_df, skeleton_df, time_range, available_events = self._eagerly_parse_lines(
              lines=lines,
              skeleton_regex=skeleton_regex,
              event_parsers=event_parsers,
              events=events,
          )
  
          self._events_df = events_df
          # If the time range is not needed, the time range we computed might be
          # wrong, so just remove it
          self._time_range = time_range if 'time-range' in self._needed_metadata else None
          self._skeleton_df = skeleton_df
          self._available_events = available_events
  
          inferred_event_descs = self._get_event_descs(skeleton_df, events, event_parsers)
          # We only needed the fields to infer the descriptors, so let's drop
          # them to lower peak memory usage
          with contextlib.suppress(KeyError):
              del self._skeleton_df['__fields']
  
          event_parsers = {
              **{
                  event: default_event_parser_cls(
                      event=event,
                      **desc,
                  )
                  for event, desc in inferred_event_descs.items()
              },
              # Existing parsers take precedence so the user can override
              # autodetected events
              **event_parsers,
          }
          self._event_parsers = event_parsers
  
  
      @classmethod
      def _resolve_event_parsers(cls, event_parsers, default_event_parser_cls):
          default_event_parser_cls = default_event_parser_cls or cls.DEFAULT_EVENT_PARSER_CLS
          event_parsers = {
              **{
                  event: (
                      desc
                      if isinstance(desc, EventParserBase)
                      else default_event_parser_cls(event=event, **desc)
                  )
                  for event, desc in cls.EVENT_DESCS.items() or []
              },
              **{
                  parser.event: parser
                  for parser in event_parsers or []
              }
          }
  
          return (default_event_parser_cls, event_parsers)
  
  
      @classmethod
      def from_string(cls, txt, **kwargs):
          """
          Build an instance from a single multiline string.
  
          :param txt: String containing the trace. It will be encoded as ASCII
              before being forwarded to the constructor if it's not already
              :class:`bytes`.
          :type txt: bytes or str
  
          :Variable keyword arguments: Forwarded to ``__init__``
          """
          # The text could already be bytes, in which case this will fail
          with contextlib.suppress(AttributeError):
              txt = txt.encode('ascii')
  
          return cls(lines=txt.splitlines(), **kwargs)
  
      @classmethod
      def from_txt_file(cls, path, **kwargs):
          """
          Build an instance from a path to a text file.
  
          :Variable keyword arguments: Forwarded to ``__init__``
          """
          with open(path, 'rb') as f:
              return cls(lines=f, **kwargs)
  
      @abc.abstractmethod
      def _get_skeleton_regex(self, need_fields):
          """
          Return a :class:`bytes` regex that provides the following groups:
  
              * ``__event``: name of the event
              * ``__time``: timestamp of event occurence
              * ``__fields`` if ``need_fields == True``: a string containing all
                the named fields of each event occurence.
  
          .. note:: This regex is critical for performances as it will be used to
              scan the whole trace.
          """
  
      @staticmethod
      def _make_df_from_data(regex, data, extra_cols=[]):
          columns = sorted(
              regex.groupindex.keys(),
              # Order columns so that we can directly append the
              # groups() tuple of the regex match
              key=lambda field: regex.groupindex[field]
          )
          # Rename regex columns to avoid clashes that were explicitly added as
          # extra
          columns = [
              '__parsed_{}'.format(col) if col in extra_cols else col
              for col in columns
          ]
          columns += extra_cols
  
          index = '__time' if data and '__time' in columns else None
          return pd.DataFrame.from_records(
              data,
              columns=columns,
              index=index,
          )
  
      def _eagerly_parse_lines(self, lines, skeleton_regex, event_parsers, events, time=None):
          """
          Filter the lines to select the ones with events.
  
          Also eagerly parse events from them to avoid the extra memory
          consumption from line storage, and to speed up parsing by acting as a
          pipeline on lazy lines stream.
          """
  
          # Recompile all regex so that they work on bytes rather than strings.
          # This simplifies the rest of the code while allowing the raw output
          # from a process to be fed
          def encode(string):
              return string.encode('ascii')
  
          events = list(map(encode, events))
          event_parsers = {
              encode(event): parser
              for event, parser in event_parsers.items()
          }
  
          # Only add an extra iterator and tuple unpacking if that is strictly
          # necessary, as it comes with a performance cost
          time_is_provided = time is not None
          skel_search = skeleton_regex.search
          if time_is_provided:
              lines = zip(time, lines)
              drop_filter = lambda line: not skel_search(line[1])
          else:
              drop_filter = lambda line: not skel_search(line)
  
          # First, get rid of all the lines coming before the trace
          lines = itertools.dropwhile(drop_filter, lines)
  
          # Appending to lists is amortized O(1). Inside the list, we store
          # tuples since they are:
          # 1) the most compact Python representation of a product type
          # 2) output directly by regex.search()
          skeleton_data = []
          events_data = {
              **{event: (None, None) for event in events},
              **{
                  event: (parser.bytes_regex.search, [])
                  for event, parser in event_parsers.items()
              },
          }
          available_events = set()
  
          begin_time = None
          end_time = None
          time_type = getattr(np, self.HEADER_FIELDS['__time'])
  
          # THE FOLLOWING LOOP IS A THE MOST PERFORMANCE-SENSITIVE PART OF THAT
          # CLASS, APPLY EXTREME CARE AND BENCHMARK WHEN MODIFYING
          # Best practices:
          # - resolve all dotted names ahead of time
          # - minimize the amount of local variables. Prefer anonymous
          #   expressions
          # - Catch exceptions for exceptional cases rather than explicit check
  
          # Pre-lookup methods out of the loop to speed it up
          append = list.append
          group = self._RE_MATCH_CLS.group
          groups = self._RE_MATCH_CLS.groups
          nextafter = np.nextafter
          inf = math.inf
          line_time = 0
          parse_time = '__time' in skeleton_regex.groupindex.keys()
  
          for line in lines:
              prev_time = line_time
              if time_is_provided:
                  line_time, line = line
  
              match = skel_search(line)
              # Stop at the first non-matching line
              try:
                  event = group(match, '__event')
                  line_time = time_type(group(match, '__time'))
              # Assume only "time" is not in the regex. Keep that out of the hot
              # path since it's only needed in rare cases (like nesting parsers)
              except IndexError:
                  # If we are supposed to parse time, let's re-raise the
                  # exception
                  if parse_time:
                      raise
                  else:
                      # Otherwise, make sure "event" is defined so that we only
                      # go a match failure on "time"
                      event
              # The line did not match the skeleton regex, so skip it
              except TypeError:
                  continue
  
              # Do a global deduplication of timestamps, across all
              # events regardless of the one we will parse. This ensures
              # stable results and joinable dataframes from multiple
              # parser instance.
              if line_time <= prev_time:
                  line_time = nextafter(prev_time, inf)
  
              if begin_time is None:
                  begin_time = line_time
  
              # If we can parse it right away, let's do it now
              try:
                  search, data = events_data[event]
                  append(
                      data,
                      # Add the fixedup time
                      groups(search(line)) + (line_time,)
                  )
              # If we don't have a parser for it yet (search == None),
              # just store the line so we can infer its parser later
              except TypeError:
                  # Add the fixedup time and the full line for later
                  # parsing as well
                  append(
                      skeleton_data,
                      groups(match) + (line_time, line)
                  )
              # We are not interested in that event, but we still remember the
              # pareseable events
              except KeyError:
                  available_events.add(event)
  
          # This should have been set on the first line.
          # Note: we don't raise the exception if no events were asked for, to
          # allow creating dummy parsers without any line
          if begin_time is None and events:
              raise ValueError('No lines containing events have been found')
  
          end_time = line_time
          available_events.update(
              event
              for event, (search, data) in events_data.items()
              if data
          )
  
          events_df = {}
          for event, parser in event_parsers.items():
              try:
                  # Remove the tuple data from the dict as we go, to free memory
                  # before proceeding to the next event to smooth the peak memory
                  # consumption
                  _, data = events_data.pop(event)
              except KeyError:
                  pass
              else:
                  decoded_event = event.decode('ascii')
                  df = self._make_df_from_data(parser.regex, data, ['__time'])
                  # Post-process immediately to shorten the memory consumption
                  # peak
                  df = self._postprocess_df(decoded_event, parser, df)
                  events_df[decoded_event] = df
  
          # Compute the skeleton dataframe for the events that have not been
          # parsed already. It contains the event name, the time, and potentially
          # the fields if they are needed
          skeleton_df = self._make_df_from_data(skeleton_regex, skeleton_data, ['__time', 'line'])
          # Drop unnecessary columns that might have been parsed by the regex
          to_keep = {'__event', '__fields', 'line'}
          skeleton_df = skeleton_df[sorted(to_keep & set(skeleton_df.columns))]
          # Make the event column more compact
          skeleton_df['__event'] = skeleton_df['__event'].astype('category', copy=False)
          # This is very fast on a category dtype
          available_events.update(skeleton_df['__event'].unique())
  
          available_events = {event.decode('ascii') for event in available_events}
          return (events_df, skeleton_df, (begin_time, end_time), available_events)
  
      def _lazyily_parse_event(self, event, parser, df):
          # Only parse the lines that have a chance to match
          df = df[df['__event'] == event.encode('ascii')]
          index = df.index
          regex = parser.bytes_regex
  
          # Resolve names outside of the comprehension
          search = regex.search
          groups = self._RE_MATCH_CLS.groups
          # Parse the lines with the regex.
          # note: we cannot use Series.str.extract(expand=True) since it does
          # not work on bytes
          data = [
              groups(search(line))
              for line in df['line']
          ]
          df = self._make_df_from_data(regex, data)
          df.index = index
          df = self._postprocess_df(event, parser, df)
          return df
  
      @staticmethod
      def _get_event_descs(df, events, event_parsers):
          user_supplied = event_parsers.keys()
          all_events = events is None
  
          if not all_events and set(events) == user_supplied:
              return {}
          else:
              def encode(string):
                  return string.encode('ascii')
  
              user_supplied = set(map(encode, user_supplied))
              events = set(map(encode, events))
  
              # Since we make a shallow copy:
              # DO NOT MODIFY ANY EXISTING COLUMN
              df = df.copy(deep=False)
  
              # Find the field names only for the events we don't already know about,
              # since the inference is relatively expensive
              if all_events:
                  df = df[~df['__event'].isin(user_supplied)]
              else:
                  events = set(events) - user_supplied
                  df = df[df['__event'].isin(events)]
  
              def apply_regex(series, func, regex):
                  regex = encode(regex.format(**TxtEventParser.PARSER_REGEX_TERMINALS))
                  regex = re.compile(regex, flags=re.ASCII)
                  return [
                      func(regex, x)
                      for x in series
                  ]
  
              def is_match(pat, x):
                  return bool(pat.search(x))
  
              # We cannot use Series.str.(count|finall) since they don't work
              # with bytes
              df['split_fields'] = apply_regex(df['__fields'], re.findall, r'({identifier})=')
              # Look for any value before the first named field
              df['pos_fields'] = apply_regex(df['__fields'], is_match, r'(?:^ *(?:[^ =]+ [^ =]+)+=)|(?:^[^=]*$)')
  
              def infer_fields(x):
                  field_alternatives = x.transform(tuple).unique()
                  if len(field_alternatives) == 1:
                      fields = field_alternatives[0]
                  else:
                      # Use the union of all the fields, and the order of appearance is not guaranteed
                      fields = sorted(set(itertools.chain.from_iterable(field_alternatives)))
  
                  return fields
  
              # For each event we don't already know, get the list of all fields available
              inferred = df.groupby('__event', observed=True)['split_fields'].apply(infer_fields).to_frame()
              inferred['positional_field'] = df.groupby('__event', observed=True)['pos_fields'].any()
  
              def update_desc(desc):
                  new = dict(
                      positional_field='__positional' if desc['positional_field'] else None,
                      fields={
                          field.decode('ascii'): None
                          for field in desc['split_fields']
                      },
                  )
                  return new
  
              dct = {
                  event.decode('ascii'): update_desc(desc)
                  for event, desc in inferred.T.to_dict().items()
              }
              return dct
  
      def parse_event(self, event):
          try:
              parser = self._event_parsers[event]
          except KeyError:
              raise MissingTraceEventError([event])
  
          # Maybe it was eagerly parsed
          try:
              df = self._events_df[event]
          except KeyError:
              df = self._lazyily_parse_event(event, parser, self._skeleton_df)
  
          # Since there is no way to distinguish between no event entry and
          # non-collected events in text traces, map empty dataframe to missing
          # event
          if df.empty:
              raise MissingTraceEventError([event])
          else:
              return df
  
      @classmethod
      def _postprocess_df(cls, event, parser, df):
          """
          ALL THE PROCESSING MUST HAPPEN INPLACE on the dataframe
          """
          # Convert fields from extracted strings to appropriate dtype
          all_fields = {
              **parser.fields,
              **cls.HEADER_FIELDS,
          }
  
          def default_converter(x):
              for dtype in cls.DTYPE_INFERENCE_ORDER:
                  convert = make_converter(dtype)
                  with contextlib.suppress(ValueError, TypeError):
                      return convert(x)
  
              # If all conversions failed, just return the initial series
              return x
  
          def make_converter(dtype):
              # If the dtype is already known, just use that
              if dtype:
                  return lambda x: series_convert(x, dtype)
              else:
                  # Otherwise, infer it from the data we have
                  return default_converter
  
          converters = {
              field: make_converter(dtype)
              for field, dtype in all_fields.items()
              if field in df.columns
          }
          # DataFrame.apply() can lead to recursion error when a conversion
          # fails, so use an explicit loop instead
          for col in set(df.columns) & converters.keys():
              df[col] = converters[col](df[col])
  
          df.index.name = 'Time'
          df.name = event
          return df
  
      def get_metadata(self, key):
          # Only return the time-range if it was asked for, as it will otherwise
          # not match the beginning of the trace
          if key == 'time-range' and key in self._needed_metadata:
              return self._time_range
          # If we filtered some events, we are not exhaustive anymore so we
          # cannot return the list
          if key == 'available-events' and not self._pre_filtered_lines:
              return self._available_events
          else:
              try:
                  return self._pre_filled_metadata[key]
              except KeyError:
                  return super().get_metadata(key)
  
  class TxtTraceParser(TxtTraceParserBase):
      """
      Text trace parser for the raw output of ``trace-cmd report -R trace.dat``.
  
      :param lines: Iterable of text lines.
      :type lines: collections.abc.Iterable(str)
  
      :param events: List of events that will be available using
          :meth:`~TxtTraceParserBase.parse_event`. If not provided, all events will be considered.
          .. note:: Restricting the set of events can speed up some operations.
      :type events: list(str)
  
      :param event_parsers: Pre-built event parsers. Missing event parsers will
          be inferred from the fields parsed in the trace, which is costly and can
          lead to using larger dtypes than necessary (e.g. ``int64`` rather than
          ``uint16``).
  
          .. seealso:: :class:`TxtEventParser`
      :type event_parsers: list(EventParserBase)
  
      :param default_event_parser_cls: Class used to build event parsers inferred from the trace.
      """
      DEFAULT_EVENT_PARSER_CLS = TxtEventParser
  
      _KERNEL_DTYPE = {
          'timestamp': 'uint64',
          'cpu': 'uint16',
          'pid': 'uint32',
          'comm': 'string',
          'cgroup_path': 'string',
          # prio in [-1, 140]
          'prio': 'int16',
          'util': 'uint16',
      }
  
      EVENT_DESCS = {
          'print': PrintTxtEventParser(
              event='print',
              func_field='func',
              content_field='buf',
          ),
          'bprint': PrintTxtEventParser(
              event='bprint',
              func_field='func',
              content_field='buf',
          ),
          'bputs': PrintTxtEventParser(
              event='bputs',
              func_field='func',
              content_field='str',
          ),
  
          'ipi_entry': dict(
              fields={
                  'reason': 'string',
              },
              positional_field='reason',
              # const char* reason is not displayed properly in raw mode
              raw=False,
          ),
          'ipi_exit': dict(
              fields={
                  'reason': 'string',
              },
              positional_field='reason',
              # const char* reason is not displayed properly in raw mode
              raw=False,
          ),
          'ipi_raise': CustomFieldsTxtEventParser(
              event='ipi_raise',
              fields_regex=r'target_mask=(?P<target_cpus>[0-9,]+) +\((?P<reason>[^)]+)\)',
              fields={
                  'target_cpus': 'string',
                  'reason': 'string',
              },
              raw=False,
          ),
          'sched_switch': dict(
              fields={
                  'prev_comm': _KERNEL_DTYPE['comm'],
                  'prev_pid': _KERNEL_DTYPE['pid'],
                  'prev_prio': _KERNEL_DTYPE['prio'],
                  'prev_state': 'uint16',
                  'next_comm': _KERNEL_DTYPE['comm'],
                  'next_pid': _KERNEL_DTYPE['pid'],
                  'next_prio': _KERNEL_DTYPE['prio'],
              },
          ),
          'sched_wakeup': dict(
              fields={
                  'comm': _KERNEL_DTYPE['comm'],
                  'pid': _KERNEL_DTYPE['pid'],
                  'prio': _KERNEL_DTYPE['prio'],
                  'target_cpu': _KERNEL_DTYPE['cpu'],
                  # This field does exist but it's useless nowadays as it has a
                  # constant value of 1, so save some memory by just not parsing
                  # it
                  # 'success': 'bool',
              },
          ),
          'cpu_frequency': dict(
              fields={
                  'cpu_id': _KERNEL_DTYPE['cpu'],
                  'state': 'uint32',
              },
          ),
          'cpu_idle': dict(
              fields={
                  'cpu_id': _KERNEL_DTYPE['cpu'],
                  # Technically, the value used in the kernel can go down to -1,
                  # but it ends up stored in an unsigned type (why ? ...), which
                  # means we actually get 4294967295 in the event, as an unsigned
                  # value. Therefore, it's up to sanitization to fix that up ...
                  'state': 'int64',
              },
          ),
          'sched_compute_energy': dict(
              fields={
                  'comm': _KERNEL_DTYPE['comm'],
                  'dst_cpu': _KERNEL_DTYPE['cpu'],
                  'energy': 'uint64',
                  'pid': _KERNEL_DTYPE['pid'],
                  'prev_cpu': _KERNEL_DTYPE['cpu'],
              },
          ),
          'sched_pelt_cfs': dict(
              fields={
                  'cpu': _KERNEL_DTYPE['cpu'],
                  'load': _KERNEL_DTYPE['util'],
                  'path': _KERNEL_DTYPE['cgroup_path'],
                  'rbl_load': _KERNEL_DTYPE['util'],
                  'util': _KERNEL_DTYPE['util'],
              },
          ),
          'sched_pelt_se': dict(
              fields={
                  'comm': _KERNEL_DTYPE['comm'],
                  'cpu': _KERNEL_DTYPE['cpu'],
                  'load': _KERNEL_DTYPE['util'],
                  'path': _KERNEL_DTYPE['cgroup_path'],
                  'pid': _KERNEL_DTYPE['pid'],
                  'rbl_load': _KERNEL_DTYPE['util'],
                  'util': _KERNEL_DTYPE['util'],
                  'update_time': _KERNEL_DTYPE['timestamp'],
              },
          ),
          'sched_migrate_task': dict(
              fields={
                  'comm': _KERNEL_DTYPE['comm'],
                  'dest_cpu': _KERNEL_DTYPE['cpu'],
                  'orig_cpu': _KERNEL_DTYPE['cpu'],
                  'pid': _KERNEL_DTYPE['pid'],
                  'prio': _KERNEL_DTYPE['prio'],
              },
          ),
          'sched_overutilized': dict(
              fields={
                  'overutilized': 'bool',
                  'span': 'string',
              },
          ),
          'sched_pelt_dl': dict(
              fields={
                  'cpu': _KERNEL_DTYPE['cpu'],
                  'load': _KERNEL_DTYPE['util'],
                  'rbl_load': _KERNEL_DTYPE['util'],
                  'util': _KERNEL_DTYPE['util'],
              },
          ),
          'sched_pelt_irq': dict(
              fields={
                  'cpu': _KERNEL_DTYPE['cpu'],
                  'load': _KERNEL_DTYPE['util'],
                  'rbl_load': _KERNEL_DTYPE['util'],
                  'util': _KERNEL_DTYPE['util'],
              },
          ),
          'sched_pelt_rt': dict(
              fields={
                  'cpu': _KERNEL_DTYPE['cpu'],
                  'load': _KERNEL_DTYPE['util'],
                  'rbl_load': _KERNEL_DTYPE['util'],
                  'util': _KERNEL_DTYPE['util'],
              },
          ),
          'sched_process_wait': dict(
              fields={
                  'comm': _KERNEL_DTYPE['comm'],
                  'pid': _KERNEL_DTYPE['pid'],
                  'prio': _KERNEL_DTYPE['prio'],
              },
          ),
          'sched_util_est_cfs': dict(
              fields={
                  'cpu': _KERNEL_DTYPE['cpu'],
                  'path': _KERNEL_DTYPE['cgroup_path'],
                  'enqueued': _KERNEL_DTYPE['util'],
                  'ewma': _KERNEL_DTYPE['util'],
                  'util': _KERNEL_DTYPE['util'],
              },
          ),
          'sched_util_est_se': dict(
              fields={
                  'cpu': _KERNEL_DTYPE['cpu'],
                  'comm': _KERNEL_DTYPE['comm'],
                  'pid': _KERNEL_DTYPE['pid'],
                  'path': _KERNEL_DTYPE['cgroup_path'],
                  'enqueued': _KERNEL_DTYPE['util'],
                  'ewma': _KERNEL_DTYPE['util'],
                  'util': _KERNEL_DTYPE['util'],
              },
          ),
          'sched_wakeup_new': dict(
              fields={
                  'comm': _KERNEL_DTYPE['comm'],
                  'pid': _KERNEL_DTYPE['pid'],
                  'prio': _KERNEL_DTYPE['prio'],
                  'success': 'bool',
                  'target_cpu': _KERNEL_DTYPE['cpu'],
              },
          ),
          'sched_waking': dict(
              fields={
                  'comm': _KERNEL_DTYPE['comm'],
                  'pid': _KERNEL_DTYPE['pid'],
                  'prio': _KERNEL_DTYPE['prio'],
                  'success': 'bool',
                  'target_cpu': _KERNEL_DTYPE['cpu'],
              },
          ),
          'uclamp_util_cfs': dict(
              fields={
                  'cpu': _KERNEL_DTYPE['cpu'],
                  'uclamp_avg': _KERNEL_DTYPE['util'],
                  'uclamp_max': _KERNEL_DTYPE['util'],
                  'uclamp_min': _KERNEL_DTYPE['util'],
                  'util_avg': _KERNEL_DTYPE['util'],
              },
          ),
          'uclamp_util_se': dict(
              fields={
                  'comm': _KERNEL_DTYPE['comm'],
                  'cpu': _KERNEL_DTYPE['cpu'],
                  'pid': _KERNEL_DTYPE['pid'],
                  'uclamp_avg': _KERNEL_DTYPE['util'],
                  'uclamp_max': _KERNEL_DTYPE['util'],
                  'uclamp_min': _KERNEL_DTYPE['util'],
                  'util_avg': _KERNEL_DTYPE['util'],
              },
          ),
      }
  
      @classmethod
      def from_dat(cls, path, events, needed_metadata=None, event_parsers=None, default_event_parser_cls=None, **kwargs):
          """
          Build an instance from a path to a trace.dat file created with
          ``trace-cmd``.
  
          :Variable keyword arguments: Forwarded to ``__init__``
          """
          needed_metadata = set(needed_metadata or [])
          events = set(events)
          # We need all the events to detect the timestamp of the first and last
          # event
          if 'time-range' in needed_metadata:
              event_filters = []
          else:
              # First get the list of events this kernel can contain, since
              # filtering on non-existant events will make the whole trace-cmd
              # report command abort.
              kernel_events = {
                  event.split(':', 1)[1]
                  for event in subprocess.check_output(
                      ['trace-cmd', 'report', '-E', '--', path],
                      stderr=subprocess.DEVNULL,
                  ).decode('ascii').splitlines()
              }
              # Filtering on event name from trace-cmd can speed up a lot the
              # partial parsing
              event_filters = list(itertools.chain.from_iterable(
                  ('-F', event)
                  for event in (events & kernel_events)
              ))
  
          default_event_parser_cls, event_parsers = cls._resolve_event_parsers(event_parsers, default_event_parser_cls)
          event_parsers = event_parsers.values()
  
          non_raw = {
              parser.event
              for parser in event_parsers
              if not parser.raw
          }
  
          raw_events = list(itertools.chain.from_iterable(
              ('-r', event) if event not in non_raw else []
              for event in events
          ))
  
          pre_filled_metadata = {}
  
          no_filter_cmd = ['trace-cmd', 'report', '-N', '-f', '--', path]
  
          if 'symbols-address' in needed_metadata:
              # Get the symbol addresses in that trace
              def parse(line):
                  addr, sym = line.split(' ', 1)
                  addr = int(addr, base=16)
                  sym = sym.strip()
                  return (addr, sym)
  
              symbols_address = dict(
                  parse(line)
                  for line in subprocess.check_output(
                      no_filter_cmd,
                      stderr=subprocess.DEVNULL,
                      universal_newlines=True,
                  ).splitlines()
              )
  
              # If we get only "0" as a key, that means kptr_restrict was in use and
              # no useable address is available
              if symbols_address.keys() != {0}:
                  pre_filled_metadata['symbols-address'] = symbols_address
  
          if 'cpus-count' in needed_metadata:
              regex = re.compile(rb'cpus=(?P<cpus>\d+)')
              with subprocess.Popen(
                  no_filter_cmd,
                  stdout=subprocess.PIPE,
                  stderr=subprocess.DEVNULL,
              ) as p:
                  try:
                      match, _ = take(1, itertools.filterfalse(None, map(regex.search, p.stdout)))
                  except ValueError:
                      pass
                  else:
                      pre_filled_metadata['cpus-count'] = match.group('cpus')
  
          kwargs.update(
              events=events,
              needed_metadata=needed_metadata,
              event_parsers=event_parsers,
              default_event_parser_cls=default_event_parser_cls,
              pre_filled_metadata=pre_filled_metadata,
          )
          cmd = [
              'trace-cmd',
              'report',
              # Do not load any plugin, so that we get fully reproducible results
              '-N',
              # Full accuracy on timestamp
              '-t',
              # All events in raw format
              *raw_events,
              *event_filters,
              '--', path
          ]
          # A fairly large buffer reduces the interaction overhead
          bufsize = 10 * 1024 * 1024
          try:
              with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=bufsize) as p:
                  # Consume the lines as they come straight from the stdout object to
                  # avoid the memory overhead of storing the whole output in one
                  # gigantic string
                  return cls(lines=p.stdout, pre_filtered_lines=bool(event_filters), **kwargs)
          except ValueError as e:
              # Maybe there were no lines in the output because no event matched the
              # filter
              if event_filters:
                  _kwargs = copy.copy(kwargs)
                  del _kwargs['events']
                  # Make a dummy parser that will raise MissingTraceEventError
                  # for any event, since the pre_filtering lead to everything
                  # being discarded
                  return cls(lines=[], events=[], pre_filtered_lines=True, **_kwargs)
              else:
                  raise
  
      def _get_skeleton_regex(self, need_fields):
          regex = r'\] +(?P<__time>{floating}): *(?P<__event>{identifier}):'.format(**TxtEventParser.PARSER_REGEX_TERMINALS)
          if need_fields:
              regex += r' *(?P<__fields>.*)'
  
          return re.compile(regex.encode('ascii'), flags=re.ASCII)
  
  
  class SimpleTxtTraceParser(TxtTraceParserBase):
      """
      Simple text trace parser (base) class.
  
      :param lines: Lines of the text to parse.
      :type lines: collections.abc.Iterable(str)
  
      :param events: List of events that will be potentially parsed by
          :meth:`~TraceParserBase.parse_events`. If ``None``, all available
          will be considered but that may increase the initial parsing stage.
      :type events: list(str)
  
      :param event_parsers: Optional list of :class:`TxtTraceParserBase` to
          provide fully customized event regex.
  
          .. note:: See :data:`EVENT_DESCS` for class-provided special case
              handling.
      :type event_parsers: list(TxtTraceParserBase)
  
      :param header_regex: Regex used to parse the header of each event. See
          :data:`HEADER_REGEX` documentation.
      :type header_regex: str
  
      .. note:: This class is easier to customize than :class:`TxtTraceParser`
          but may have higher processing time and peak memory usage.
      """
  
      EVENT_DESCS = {}
      """
      Mapping of event names to parser description as a dict.
  
      Each event description can include the following dict keys:
  
          * ``header_regex``: Regex to parse the event header. If not set, the
            header regex from the trace parser will be used.
  
          * ``fields_regex``: Regex to parse the fields part of the event (i.e.
            the part after the header). This is the most commonly modified
            setting to take into account special cases in event formatting.
  
          * ``fields``: Mapping of field names to :class:`pandas.DataFrame`
            column dtype. This allows using a smaller dtype or the use of a
            non-inferred dtype like ``boolean``.
  
          * ``positional_field``: Name of the positional field (comming before
            the named fields). If ``None``, the column will be suppressed in the
            parsed dataframe.
      """
  
      HEADER_REGEX = None
      """
      Default regex to use to parse event header.
      It must parse the following groups:
  
          * ``__time``: the timestamp of the event
          * ``__event``: the name of the event
          * ``__cpu`` (optional): the CPU by which the event was emitted
          * ``__pid`` (optional): the currently scheduled PID at the point the event was emitted
          * ``__comm`` (optional): the currently scheduled task's name at the point the event was emitted
  
      .. note:: It must *not* capture the event fields, as it will be
          concatenated with the field regex of each event to parse full lines.
      """
  
      def __init__(self, lines, events=None, event_parsers=None, header_regex=None, **kwargs):
          header_regex = header_regex or self.HEADER_REGEX
          self.header_regex = header_regex
  
          # Do not parse for each event unnecessary columns that are already
          # parsed in the skeleton dataframe
          regex = header_regex
          for field in ('__time', '__event'):
              regex = regex.replace(r'(?P<{}>'.format(field), r'(?:')
          event_parser_header_regex = regex
  
          class SimpleTxtEventParser(TxtEventParser):
              def __init__(self, header_regex=event_parser_header_regex, fields_regex=None, **kwargs):
                  self.fields_regex = fields_regex
                  self.header_regex = header_regex
                  super().__init__(**kwargs)
  
              def _get_header_regex(self, event):
                  return self.header_regex
  
              def _get_fields_regex(self, *args, **kwargs):
                  if self.fields_regex:
                      return self.fields_regex
                  else:
                      return super()._get_fields_regex(*args, **kwargs)
  
          super().__init__(
              lines=lines,
              events=events,
              event_parsers=event_parsers,
              default_event_parser_cls=SimpleTxtEventParser,
              **kwargs,
          )
  
      def _get_skeleton_regex(self, need_fields):
          # Parse the whole header, which is wasteful but provides a simpler interface
          regex = self.header_regex + r' *(?P<__fields>.*)'
          return re.compile(regex.encode('ascii'), flags=re.ASCII)
  
  
  class MetaTxtTraceParser(SimpleTxtTraceParser):
      """
      Textual trace parser to parse meta-events.
  
      :param time: Iterable of timestamps matching ``lines``.
      :type time: collections.abc.Iterable(float)
  
      :param merged_df: Dataframe to merge into the parsed ones, to add
          pre-computed fields.
      :type merged_df: pandas.DataFrame
  
      Meta events are events "embedded" as a string inside the field of another
      event. They are expected to comply with the raw format as output by
      ``trace-cmd report -R``.
  
      """
      HEADER_REGEX = r'(?P<__event>[\w@]+):?'
  
      class DEFAULT_EVENT_PARSER_CLS(TxtEventParser):
          @classmethod
          def _get_header_regex(cls, event):
              regex = r'^ *{__event}:?'.format(
                  __event=re.escape(event),
                  **cls.PARSER_REGEX_TERMINALS
              )
              return regex
  
      EVENT_DESCS = {
          **SimpleTxtTraceParser.EVENT_DESCS,
          # Provide the description of rtapp userspace events to speed up parsing
          # of these
          'userspace@rtapp_loop': dict(
              fields={
                  'phase': 'uint32',
                  'phase_loop': 'uint32',
                  'thread_loop': 'uint32',
                  'event': 'string',
              },
          ),
          'userspace@rtapp_main': dict(
              fields={
                  'event': 'string',
                  'data': None,
              },
          ),
          'userspace@rtapp_task': dict(
              fields={
                  'event': 'string',
              },
          ),
          'userspace@rtapp_event': dict(
              fields={
                  'desc': 'string',
                  # Not sure about the meaning or real size of these
                  'id': 'uint32',
                  'type': 'uint32',
              },
          ),
          'userspace@rtapp_stats': dict(
              fields={
                  'period': 'uint64',
                  'run': 'uint64',
                  'wa_lat': 'uint64',
                  'slack': 'uint64',
                  'c_period': 'uint64',
                  'c_run': 'uint64',
              },
          ),
      }
  
      def __init__(self, *args, time, merged_df=None, **kwargs):
          self._time = time
          self._merged_df = merged_df
          super().__init__(*args, **kwargs)
  
      def _eagerly_parse_lines(self, *args, **kwargs):
          # Use the iloc as "time", and we fix it up manually afterwards
          return super()._eagerly_parse_lines(
              *args, **kwargs, time=self._time,
          )
  
      def _postprocess_df(self, event, parser, df):
          df = super()._postprocess_df(event, parser, df)
          merged_df = self._merged_df
          if merged_df is not None:
              df = df.merge(merged_df, left_index=True, right_index=True, copy=False)
          return df
  
  class HRTxtTraceParser(SimpleTxtTraceParser):
      """
      Parse text trace in their human readable format (as opposed to the raw
      format).
  
      The accepted format is the one produced by the kernel, after formatting
      event records with their format string. This means that format strings
      deviating from the classic ``field=value`` format need a custom regex.
  
      .. note:: This parser is provided for convenience but is probably not
          complete. More specifically, it does not contain custom regex for all the
          events which format deviates from the raw format as output by ``trace-cmd
          report -R``.
  
          For a better supported format, see :class:`TxtTraceParser`.
      """
      EVENT_DESCS = {
          'sched_switch': dict(
              fields_regex=r'prev_comm=(?P<prev_comm>.+?) +prev_pid=(?P<prev_pid>\d+) +prev_prio=(?P<prev_prio>\d+) +prev_state=(?P<prev_state>[^ ]+) ==> next_comm=(?P<next_comm>.+?) +next_pid=(?P<next_pid>\d+) +next_prio=(?P<next_prio>\d+)',
              fields={
                  'prev_comm': 'string',
                  'prev_pid': 'uint32',
                  'prev_prio': 'int16',
                  'prev_state': 'string',
                  'next_comm': 'string',
                  'next_pid': 'uint32',
                  'next_prio': 'int16',
              },
          ),
          'tracing_mark_write': dict(
              fields={
                  'buf': 'string',
              },
              positional_field='buf',
          ),
      }
  
      HEADER_REGEX = r'(?P<__comm>\S+)-(?P<__pid>\d+).+(?P<__time>\d+\.\d+): +(?P<__event>\w+):'
  
  
  class SysTraceParser(HRTxtTraceParser):
      """
      Parse Google's systrace format.
  
      .. note:: This parser is based on :class:`HRTxtTraceParser` and is
          therefore provided for convenience but may lack some events custom
          field regex.
      """
  
      @classmethod
      def from_html(cls, *args, **kwargs):
          return super().from_txt_file(*args, **kwargs)
  
  
  class TrappyTraceParser(TraceParserBase):
      """
      Glue with :mod:`trappy` trace parsers.
  
      .. note:: This class is deprecated and the use of :class:`TxtTraceParser`
          is recommended instead. This class is only provided in case
          compatibility with :mod:`trappy` is needed for one reason or another.
      """
  
      def __init__(self, path, events, needed_metadata=None, trace_format=None):
          super().__init__(events, needed_metadata=needed_metadata)
  
          # Lazy import so that it's not a required dependency
          import trappy
  
          events = set(events)
          # Make sure we won't attempt parsing 'print' events, so that this
          # parser is not used to resolve meta events.
          events -= {'print'}
          self._events = events
  
          events = set(events)
          if trace_format is None:
              if path.endswith('html'):
                  trace_format = 'SySTrace'
              else:
                  trace_format = 'FTrace'
  
          self.get_logger().debug('Parsing {} events from {}: {}'.format(trace_format, path, sorted(events)))
          if trace_format == 'SysTrace':
              trace_class = trappy.SysTrace
          elif trace_format == 'FTrace':
              trace_class = trappy.FTrace
          else:
              raise ValueError('Unknown trace format: {}'.format(trace_format))
  
          # Since we handle the cache in lisa.trace.Trace, we do not need to duplicate it
          trace_class.disable_cache = True
          trace = trace_class(
              path,
              scope="custom",
              events=sorted(events),
              normalize_time=False,
          )
          self._trace = trace
  
          # trappy sometimes decides to be "clever" and overrules the path to be
          # used, even though it was specifically asked for a given file path
          assert path == trace.trace_path
  
      def get_metadata(self, key):
          if key == 'time-range':
              trace = self._trace
              return (trace.basetime, trace.endtime)
          else:
              return super().get_metadata(key)
  
      def parse_event(self, event):
          trace = self._trace
  
          if event not in self._events:
              raise MissingTraceEventError([event])
  
          df = getattr(trace, event).data_frame
          if df.empty:
              raise MissingTraceEventError([event])
  
          return df
  
  
  class TraceBase(abc.ABC):
      """
      Base class for common functionalities between :class:`Trace` and :class:`TraceView`
      """
  
      def __init__(self):
          # Import here to avoid a circular dependency issue at import time
          # with lisa.analysis.base
          from lisa.analysis.proxy import AnalysisProxy
          self.analysis = AnalysisProxy(self)
  
      @property
      def trace_state(self):
          """
          State of the trace object that might impact the output of dataframe
          getter functions like :meth:`Trace.df_event`.
  
          It must be hashable and serializable to JSON, so that it can be
          recorded when analysis methods results are cached to the swap.
          """
          return None
  
      @property
      def time_range(self):
          """
          Duration of that trace.
          """
          return self.end - self.start
  
      @property
      def window(self):
          """
          Same as ``(trace.start, trace.end)``.
  
          This is handy to pass to functions expecting a window tuple.
          """
          return (self.start, self.end)
  
      @abc.abstractmethod
      def get_view(self, window, **kwargs):
          """
          Get a view on a trace cropped time-wise to fit in ``window``
  
          :Variable keyword arguments: Forwarded to the contructor of the view.
          """
          pass
  
      def __getitem__(self, window):
          if not isinstance(window, slice):
              raise TypeError("Cropping window must be an instance of slice")
  
          if window.step is not None:
              raise ValueError("Slice step is not supported")
  
          return self.get_view((window.start, window.stop))
  
      @deprecate('Prefer adding delta once signals have been extracted from the event dataframe for correctness',
          deprecated_in='2.0',
          removed_in='2.1',
          replaced_by=df_add_delta,
      )
      def add_events_deltas(self, df, col_name='delta', inplace=True):
          """
          Store the time between each event in a new dataframe column
  
          This function assumes that at time [n] the event starts and at [n+1]
          the event stops, so the formula for the returned value is::
  
                    |        |            |
                    |        |            |
                    |        |            |
              ------+--------+------------+------
                  [n-1]     [n]         [n+1]
  
              delta[n] = index[n+1] - index[n]
  
          :param df: The DataFrame to operate one
          :type df: pandas.DataFrame
  
          :param col_name: The name of the column to add
          :type col_name: str
  
          :param inplace: Whether to operate on the passed DataFrame, or to use
            a copy of it
          :type inplace: bool
  
          This method only really makes sense for events tracking an on/off state
          (e.g. overutilized, idle)
          """
  
          if col_name in df.columns:
              raise RuntimeError("Column {} is already present in the dataframe".
                                 format(col_name))
  
          return df_add_delta(df, col=col_name, inplace=inplace, window=self.window)
  
      @deprecate('This method has been deprecated and is an alias',
          deprecated_in='2.0',
          removed_in='2.1',
          replaced_by='df_event',
      )
      def df_events(self, *args, **kwargs):
          return self.df_event(*args, **kwargs)
  
      def df_all_events(self, events=None):
          """
          Provide a dataframe with an ``info`` column containing the textual
          human-readable representation of the events fields.
  
          :param events: List of events to include. If ``None``, all parsed
              events will be used.
          :type events: list(str) or None
          """
          if events is None:
              events = sorted(self.available_events)
  
          if not events:
              return pd.DataFrame({'info': []})
  
          max_event_name_len = max(len(event) for event in events)
  
          def make_info_row(row, event):
              fields = ' '.join(
                  '{}={}'.format(key, value)
                  for key, value in row.iteritems()
              )
  
              return '{:<{event_name_len}}: {}'.format(event, fields, event_name_len=max_event_name_len)
  
          def make_info_series(event):
              df = self.df_event(event)
              info = df.apply(make_info_row, axis=1, event=event)
              info.name = 'info'
              return info
  
          series_list = [
              make_info_series(event)
              for event in events
          ]
  
          series = pd.concat(series_list)
          series.sort_index(inplace=True)
          return pd.DataFrame({'info': series})
  
  
  class TraceView(Loggable, TraceBase):
      """
      A view on a :class:`Trace`
  
      :param trace: The trace to trim
      :type trace: Trace
  
      :param clear_base_cache: Clear the cache of the base ``trace`` for non-raw
          data. This can release memory if the base trace is not going to be used
          anymore, apart as a data server for the view.
      :type clear_base_cache: bool
  
      :param window: The time window to base this view on
      :type window: tuple(float, float)
  
      :ivar base_trace: The original :class:`Trace` this view is based on.
      :ivar analysis: The analysis proxy on the trimmed down :class:`Trace`.
  
      :ivar start: The timestamp of the first trace event in the view (>= ``window[0]``)
      :ivar end: The timestamp of the last trace event in the view (<= ``window[1]``)
  
      You can substitute an instance of :class:`Trace` with an instance of
      :class:`TraceView`. This means you can create a view of a trimmed down trace
      and run analysis code/plots that will only use data within that window, e.g.::
  
        trace = Trace(...)
        view = trace.get_view((2, 4))
  
        # Alias for the above
        view = trace[2:4]
  
        # This will only use events in the (2, 4) time window
        df = view.analysis.tasks.df_tasks_runtime()
  
      **Design notes:**
  
        * :meth:`df_event` uses the underlying :meth:`lisa.trace.Trace.df_event`
          and trims the dataframe according to the given ``window`` before
          returning it.
        * ``self.start`` and ``self.end`` mimic the :class:`Trace` attributes but
          they are adjusted to match the given window. On top of this, this class
          mimics a regular :class:`Trace` using :func:`getattr`.
      """
  
      def __init__(self, trace, window, clear_base_cache=False):
          super().__init__()
          self.base_trace = trace
  
          # evict all the non-raw dataframes from the base cache, as they are
          # unlikely to be used anymore.
          if clear_base_cache:
              self.base_trace._cache.clear_all_events(raw=False)
  
          t_min, t_max = window
          self.start = t_min if t_min is not None else self.base_trace.start
          self.end = t_max if t_max is not None else self.base_trace.end
  
      @property
      def trace_state(self):
          return (self.start, self.end, self.base_trace.trace_state)
  
      def __getattr__(self, name):
          return getattr(self.base_trace, name)
  
      def df_event(self, event, **kwargs):
          """
          Get a dataframe containing all occurrences of the specified trace event
          in the sliced trace.
  
          :param event: Trace event name
          :type event: str
  
          :Variable keyword arguments: Forwarded to
              :meth:`lisa.trace.Trace.df_event`.
          """
          try:
              window = kwargs['window']
          except KeyError:
              window = (self.start, self.end)
          kwargs['window'] = window
  
          return self.base_trace.df_event(event, **kwargs)
  
      def get_view(self, window, **kwargs):
          start = self.start
          end = self.end
  
          if window[0]:
              start = max(start, window[0])
  
          if window[1]:
              end = min(end, window[1])
  
          return self.base_trace.get_view(window=(start, end), **kwargs)
  
  # One might be tempted to make that a subclass of collections.abc.Set: don't.
  # The problem is that Set expects new instances to be created by passing an
  # iterable to __init__(), but that container cannot be randomly instanciated
  # with values, it is tied to a Trace.
  class _AvailableTraceEventsSet:
      """
      Smart container that uses demand event loading on the trace to check
      whether an event is present or not.
  
      This container can be iterated over to get the current available events,
      and supports membership tests using ``in``.
      """
      def __init__(self, trace):
          self._trace = trace
  
      def __contains__(self, event):
          trace = self._trace
  
          if trace._strict_events and not trace._is_meta_event(event):
              return trace._parseable_events.setdefault(event, False)
  
          # Try to parse the event in case it was not parsed already
          if event not in trace._parseable_events:
              # If the trace file is not accessible anymore, we will get an OSError
              with contextlib.suppress(MissingTraceEventError, OSError):
                  trace.df_event(event=event, raw=True)
  
          return trace._parseable_events.setdefault(event, False)
  
      @property
      def _available_events(self):
          return {
              event
              for event, available in self._trace._parseable_events.items()
              if available
          }
  
      def __iter__(self):
          return iter(self._available_events)
  
      def __len__(self):
          return len(self._available_events)
  
      def __str__(self):
          return str(self._available_events)
  
  
  class PandasDataDesc(Mapping):
      """
      Pandas data descriptor.
  
      :param spec: Specification of the data as a key/value mapping.
  
      This holds all the information needed to uniquely identify a
      :class:`pandas.DataFrame` or :class:`pandas.Series`. It is used to manage
      the cache and swap.
  
      It implements the :class:`collections.abc.Mapping` interface, so
      specification keys can be accessed directly like from a dict.
  
      .. note:: Once introduced in a container, instances must not be modified,
          directly or indirectly.
  
      :ivar normal_form: Normal form of the descriptor. Equality is implemented
          by comparing this attribute.
      :vartype normal_form: PandasDataDescNF
      """
  
      def __init__(self, spec):
          self.spec = spec
          self.normal_form = PandasDataDescNF.from_spec(self.spec)
  
      def __getitem__(self, key):
          return self.spec[key]
  
      def __iter__(self):
          return iter(self.spec)
  
      def __len__(self):
          return len(self.spec)
  
      @classmethod
      def from_kwargs(cls, **kwargs):
          """
          Build a :class:`PandasDataDesc` with the specifications as keyword
          arguments.
          """
          return cls(spec=kwargs)
  
      def __repr__(self):
          return '{}({})'.format(
              self.__class__.__name__,
              ', '.join(
                  '{}={!r}'.format(key, val)
                  for key, val in self.__dict__.items()
              )
          )
  
      def __eq__(self, other):
          if isinstance(other, self.__class__):
              return self.normal_form == other.normal_form
          else:
              return False
  
      def __hash__(self):
          return hash(self.normal_form)
  
  
  class PandasDataDescNF:
      """
      Normal form of :class:`PandasDataDesc`.
  
      The normal form of the descriptor allows removing any possible differences
      in shape of values, and is serializable to JSON. The serialization is
      allowed to destroy some information (type mainly), as long as it does make
      two descriptors wrongly equal.
      """
      def __init__(self, nf):
          self._nf = nf
          # Since it's going to be inserted in dict for sure, precompute the hash
          # once and for all.
          self._hash = hash(self._nf)
  
      @classmethod
      def from_spec(cls, spec):
          """
          Build from a spec that can include any kind of Python objects.
          """
          nf = tuple(sorted(
              (key, cls._coerce(val))
              for key, val in spec.items()
          ))
          return cls(nf=nf)
  
      @classmethod
      def _coerce(cls, val):
          "Coerce data to a normal form that must be hashable"
          if isinstance(val, Integral):
              val = int(val)
          elif isinstance(val, Real):
              val = float(val)
          elif isinstance(val, (type(None), str, Number)):
              pass
          elif isinstance(val, Mapping):
              val = tuple(
                  (cls._coerce(key), cls._coerce(val))
                  for key, val in sorted(val.items())
              )
              val = ('mapping', val)
          elif isinstance(val, Set):
              val = tuple(map(cls._coerce, sorted(val)))
              val = ('set', val)
          elif isinstance(val, Sequence):
              val = tuple(map(cls._coerce, val))
              val = ('sequence', val)
          # In other cases save the name of the type along the value to make
          # sure we are not going to compare apple and oranges in the future
          else:
              type_name = '{}.{}'.format(
                  val.__class__.__module__,
                  val.__class__.__qualname__
              )
              val = (type_name, val)
  
          return val
  
      def __str__(self):
          return str(self._nf)
  
      def __eq__(self, other):
          if isinstance(other, self.__class__):
              return self._nf == other._nf
          else:
              return False
  
      def __hash__(self):
          return self._hash
  
      def to_json_map(self):
          return dict(self._nf)
  
      @classmethod
      def _coerce_json(cls, x):
          """
          JSON converts the original tuples into lists, so we need to convert it
          back.
          """
          if isinstance(x, str):
              return x
          elif isinstance(x, Sequence):
              return tuple(map(cls._coerce_json, x))
          else:
              return x
  
      @classmethod
      def from_json_map(cls, mapping):
          """
          Build from a mapping that was created using :meth:`to_json_map`.
  
          JSON does not preserve tuples for example, so they need to be converted
          back.
          """
          nf = tuple(sorted(
              (key, cls._coerce_json(val))
              for key, val in mapping.items()
          ))
          return cls(nf=nf)
  
  class PandasDataSwapEntry:
      """
      Entry in the pandas data swap area of :class:`Trace`.
  
      :param pd_desc_nf: Normal form descriptor describing what the entry
          contains.
      :type pd_desc_nf: PandasDataDescNF
  
      :param name: Name of the entry. If ``None``, a random UUID will be
          generated.
      :type name: str or None
      """
  
      META_EXTENSION = '.meta'
      """
      Extension used by the metadata file of the swap entry in the swap.
      """
  
      def __init__(self, pd_desc_nf, name=None):
          self.pd_desc_nf = pd_desc_nf
          self.name = name or uuid.uuid4().hex
  
      @property
      def meta_filename(self):
          """
          Filename of the metadata file in the swap.
          """
          return '{}{}'.format(self.name, self.META_EXTENSION)
  
      @property
      def data_filename(self):
          """
          Filename of the pandas data file in the swap.
          """
          return '{}{}'.format(self.name, TraceCache.DATAFRAME_SWAP_EXTENSION)
  
      def to_json_map(self):
          """
          Return a mapping suitable for JSON serialization.
          """
          return {
              'version-token': VERSION_TOKEN,
              'name': self.name,
              'desc': self.pd_desc_nf.to_json_map(),
          }
  
      @classmethod
      def from_json_map(cls, mapping):
          """
          Create an instance with a mapping created using :meth:`to_json_map`.
          """
          if mapping['version-token'] != VERSION_TOKEN:
              raise TraceCacheSwapVersionError('Version token differ')
  
          pd_desc_nf = PandasDataDescNF.from_json_map(mapping['desc'])
          name = mapping['name']
          return cls(pd_desc_nf=pd_desc_nf, name=name)
  
      def to_path(self, path):
          """
          Save the swap entry metadata to the given ``path``.
          """
          data = self.to_json_map()
          with open(path, 'w') as f:
              json.dump(data, f)
              f.write('\n')
  
      @classmethod
      def from_path(cls, path):
          """
          Load the swap entry metadata from the given ``path``.
          """
          with open(path) as f:
              mapping = json.load(f)
  
          return cls.from_json_map(mapping)
  
  
  class TraceCacheSwapVersionError(ValueError):
      """
      Exception raised when the swap entry was created by another version of LISA
      than the one loading it.
      """
      pass
  
  
  class TraceCache(Loggable):
      """
      Cache of a :class:`Trace`.
  
      :param max_mem_size: Maximum amount of memory to use in bytes. If the data
          hold in memory exceed that size, they will be evicted to the swap if
          possible and if it would be faster to reload them from swap rather than
          recomputing them. If there is no swap area, the data is just discarded.
      :type max_mem_size: int or None
  
      :param max_swap_size: Maximum amount of swap to use in bytes. When the
          amount of data saved to the swap exceeds that threshold, older files
          are discarded.
      :type max_swap_size: int or None
  
      :param swap_dir: Folder to use as swap area.
      :type swap_dir: str or None
  
      :param trace_path: Absolute path of the trace file.
      :type trace_path: str or None
  
      :param trace_md5: MD5 checksum of the trace file, to invalidate the cache
          if the file changed.
      :type trace_md5: str or None
  
      :param metadata: Metadata mapping to store in the swap area.
      :type metadata: dict or None
  
      :param swap_content: Initial content of the swap area.
      :type swap_content: dict(PandasDataDescNF, PandasDataSwapEntry) or None
  
      The cache manages both the :class:`pandas.DataFrame` and
      :class:`pandas.Series` generated in memory and a swap area used to evict
      them, and to reload them quickly.
      """
  
      INIT_SWAP_COST = 1e-7
      """
      Somewhat arbitrary number, must be small enough so that we write at
      least one dataset to the cache, which will allow us getting a better
      estimation. If the value is too high from the start, we will never
      write anything, and the value will never have a chance to re-adjust.
      """
  
      TRACE_META_FILENAME = 'trace.meta'
      """
      Name of the trace metadata file in the swap area.
      """
  
      DATAFRAME_SWAP_FORMAT = 'parquet'
      """
      Data storage format used to swap.
      """
  
      DATAFRAME_SWAP_EXTENSION = '.{}'.format(DATAFRAME_SWAP_FORMAT)
      """
      File extension of the data swap format.
      """
  
      def __init__(self, max_mem_size=None, trace_path=None, trace_md5=None, swap_dir=None, max_swap_size=None, swap_content=None, metadata=None):
          self._cache = {}
          self._data_cost = {}
          self._swap_content = swap_content or {}
          self._pd_desc_swap_filename = {}
          self.swap_cost = self.INIT_SWAP_COST
          self.swap_dir = swap_dir
          self.max_swap_size = max_swap_size if max_swap_size is not None else math.inf
          self._swap_size = self._get_swap_size()
  
          self.max_mem_size = max_mem_size if max_mem_size is not None else math.inf
          self._data_mem_swap_ratio = 7
          self._metadata = metadata or {}
  
          self.trace_path = os.path.abspath(trace_path)
          self._trace_md5 = trace_md5
  
      @property
      @memoized
      def _swap_size_overhead(self):
          def make_df(nr_col):
              return pd.DataFrame({
                  str(x): []
                  for x in range(nr_col)
              })
  
          def get_size(nr_col):
              df = make_df(nr_col)
              buffer = io.BytesIO()
              self._write_data(df, buffer)
              return buffer.getbuffer().nbytes
  
          size1 = get_size(1)
          size2 = get_size(2)
  
          col_overhead = size2 - size1
          # Since parquet seems to fail serializing of a dataframe with 0 columns
          # in some cases, we use the dataframe with one column and remove the
          # overhead of the column. It gives almost the same result.
          file_overhead = size1 - col_overhead
          assert col_overhead > 0
  
          return (file_overhead, col_overhead)
  
      def _unbias_swap_size(self, data, size):
          """
          Remove the fixed size overhead of the file format being used, assuming
          a non-compressible overhead per file and per column.
  
          .. note:: This model seems to work pretty well for parquet format.
          """
          file_overhead, col_overhead = self._swap_size_overhead
          # DataFrame
          try:
              nr_columns = data.shape[1]
          # Series
          except IndexError:
              nr_columns = 1
  
          size = size - file_overhead - nr_columns * col_overhead
          return size
  
      @property
      def trace_md5(self):
          md5 = self._trace_md5
          if md5 is None:
              with open(self.trace_path, 'rb') as f:
                  md5 = checksum(f, 'md5')
              self._trace_md5 = md5
  
          return md5
  
      def update_metadata(self, metadata):
          """
          Update the metadata mapping with the given ``metadata`` mapping and
          write it back to the swap area.
          """
          self._metadata.update(metadata)
          self.to_swap_dir()
  
      def get_metadata(self, key):
          """
          Get the value of the given metadata ``key``.
          """
          try:
              return self._metadata[key]
          except KeyError:
              raise MissingMetadataError(key)
  
      def to_json_map(self):
          """
          Returns a dictionary suitable for JSON serialization.
          """
  
          if self.swap_dir:
              trace_path = os.path.relpath(self.trace_path, self.swap_dir)
          else:
              trace_path = os.path.abspath(self.trace_path)
  
          return {
              'version-token': VERSION_TOKEN,
              'metadata': self._metadata,
              'trace-path': trace_path,
              'trace-md5': self.trace_md5,
          }
  
      def to_path(self, path):
          """
          Write the persistent state to the given ``path``.
          """
          mapping = self.to_json_map()
          with open(path, 'w') as f:
              json.dump(mapping, f)
              f.write('\n')
  
      @classmethod
      def _from_swap_dir(cls, swap_dir, trace_path=None, metadata=None, **kwargs):
          metapath = os.path.join(swap_dir, cls.TRACE_META_FILENAME)
  
          with open(metapath) as f:
              mapping = json.load(f)
  
          if mapping['version-token'] != VERSION_TOKEN:
              raise TraceCacheSwapVersionError('Version token differ')
  
          swap_trace_path = mapping['trace-path']
          swap_trace_path = os.path.join(swap_dir, swap_trace_path)
  
          metadata = metadata or {}
  
          try:
              with open(swap_trace_path, 'rb') as f:
                  new_md5 = checksum(f, 'md5')
          except FileNotFoundError:
              new_md5 = None
  
          if trace_path and not os.path.samefile(swap_trace_path, trace_path):
              invalid_swap = True
          else:
              if new_md5 is None:
                  invalid_swap = True
              else:
                  old_md5 = mapping['trace-md5']
                  invalid_swap = (old_md5 != new_md5)
  
          if invalid_swap:
              # Remove the invalid swap and create a fresh directory
              shutil.rmtree(swap_dir)
              os.makedirs(swap_dir)
              swap_content = None
          else:
              def load_swap_content(swap_dir):
                  swap_entry_filenames = {
                      filename
                      for filename in os.listdir(swap_dir)
                      if filename.endswith(PandasDataSwapEntry.META_EXTENSION)
                  }
  
                  for filename in swap_entry_filenames:
                      path = os.path.join(swap_dir, filename)
                      try:
                          swap_entry = PandasDataSwapEntry.from_path(path)
                      # If there is any issue with that entry, just ignore it
                      except Exception:
                          continue
                      else:
                          yield (swap_entry.pd_desc_nf, swap_entry)
  
              swap_content = dict(load_swap_content(swap_dir))
  
              metadata_ = mapping['metadata']
              metadata = {**metadata_, **metadata}
  
          return cls(swap_content=swap_content, swap_dir=swap_dir, metadata=metadata, trace_path=trace_path, trace_md5=new_md5, **kwargs)
  
      def to_swap_dir(self):
          """
          Write the persistent state to the swap area if any, no-op otherwise.
          """
          if self.swap_dir:
              path = os.path.join(self.swap_dir, self.TRACE_META_FILENAME)
              self.to_path(path)
  
      @classmethod
      def from_swap_dir(cls, swap_dir, **kwargs):
          """
          Reload the persistent state from the given ``swap_dir``.
  
          :Variable keyword arguments: Forwarded to :class:`TraceCache`.
          """
          if swap_dir:
              try:
                  return cls._from_swap_dir(swap_dir=swap_dir, **kwargs)
              except (FileNotFoundError, TraceCacheSwapVersionError, json.decoder.JSONDecodeError):
                  pass
  
          return cls(swap_dir=swap_dir, **kwargs)
  
      def _estimate_data_swap_cost(self, data):
          return self._estimate_data_swap_size(data) * self.swap_cost
  
      def _estimate_data_swap_size(self, data):
          return self._data_mem_usage(data) * self._data_mem_swap_ratio
  
      def _update_ewma(self, attr, new, alpha=0.25, override=False):
          old = getattr(self, attr)
          if override:
              updated = new
          else:
              updated = (1 - alpha) * old + alpha * new
  
          setattr(self, attr, updated)
  
      def _update_data_swap_size_estimation(self, data, size):
          size = self._unbias_swap_size(data, size)
  
          # If size < 0, the dataframe is so small that it's basically just noise
          if size > 0:
              mem_usage = self._data_mem_usage(data)
              if mem_usage:
                  self._update_ewma('_data_mem_swap_ratio', size / mem_usage)
  
      def _data_mem_usage(self, data):
          mem = data.memory_usage()
          try:
              return mem.sum()
          except AttributeError:
              return mem
  
      def _should_evict_to_swap(self, pd_desc, data):
          # If we don't have any cost info, assume it is expensive to compute
          compute_cost = self._data_cost.get(pd_desc, math.inf)
          swap_cost = self._estimate_data_swap_cost(data)
          return swap_cost <= compute_cost
  
      def _swap_path_of(self, pd_desc):
          if self.swap_dir:
              pd_desc_nf = pd_desc.normal_form
              swap_entry = self._swap_content[pd_desc_nf]
              filename = swap_entry.data_filename
              return os.path.join(self.swap_dir, filename)
          else:
              raise ValueError('Swap dir is not setup')
  
      def _update_swap_cost(self, data, swap_cost, mem_usage, swap_size):
          unbiased_swap_size = self._unbias_swap_size(data, swap_size)
          # Take out from the swap cost the time it took to write the overhead
          # that comes with the file format, assuming the cost is
          # proportional to amount of data written in the swap.
          swap_cost *= unbiased_swap_size / swap_size
  
          new_cost = swap_cost / mem_usage
  
          override = self.swap_cost == self.INIT_SWAP_COST
          # EWMA to keep a relatively stable cost
          self._update_ewma('swap_cost', new_cost, override=override)
  
      def _is_written_to_swap(self, pd_desc):
          return pd_desc.normal_form in self._swap_content
  
      @classmethod
      def _write_data(cls, data, path):
          if cls.DATAFRAME_SWAP_FORMAT == 'parquet':
              # Snappy compression seems very fast
              data.to_parquet(path, compression='snappy', index=True)
          else:
              raise ValueError('Dataframe swap format "{}" not handled'.format(cls.DATAFRAME_SWAP_FORMAT))
  
      def _write_swap(self, pd_desc, data):
          if not self.swap_dir:
              return
          else:
              if self._is_written_to_swap(pd_desc):
                  return
  
              pd_desc_nf = pd_desc.normal_form
              swap_entry = PandasDataSwapEntry(pd_desc_nf)
  
              df_path = os.path.join(self.swap_dir, swap_entry.data_filename)
  
              # If that would make the swap dir too large, try to do some cleanup
              if self._estimate_data_swap_size(data) + self._swap_size > self.max_swap_size:
                  self.scrub_swap()
  
              def log_error(e):
                  self.get_logger().error('Could not write {} to swap: {}'.format(pd_desc, e))
  
              # Write the Parquet file and update the write speed
              try:
                  with measure_time() as measure:
                      self._write_data(data, df_path)
              # PyArrow fails to save dataframes containing integers > 64bits
              except OverflowError as e:
                  log_error(e)
              else:
                  # Update the swap
                  swap_entry_path = os.path.join(self.swap_dir, swap_entry.meta_filename)
                  swap_entry.to_path(swap_entry_path)
                  self._swap_content[swap_entry.pd_desc_nf] = swap_entry
  
                  # Assume that reading from the swap will take as much time as
                  # writing to it. We cannot do better anyway, but that should
                  # mostly bias to keeping things in memory if possible.
                  swap_cost = measure.exclusive_delta
                  data_swapped_size = os.stat(df_path).st_size
  
                  mem_usage = self._data_mem_usage(data)
                  if mem_usage:
                      self._update_swap_cost(data, swap_cost, mem_usage, data_swapped_size)
                  self._swap_size += data_swapped_size
                  self._update_data_swap_size_estimation(data, data_swapped_size)
                  self.scrub_swap()
  
      def _get_swap_size(self):
          if self.swap_dir:
              return sum(
                  dir_entry.stat().st_size
                  for dir_entry in os.scandir(self.swap_dir)
              )
          else:
              return 0
  
      def scrub_swap(self):
          """
          Scrub the swap area to remove old files if the storage size limit is exceeded.
          """
          # TODO: Load the file information from __init__ by discovering the swap
          # area's content to avoid doing it each time here
          if self._swap_size > self.max_swap_size and self.swap_dir:
              stats = {
                  dir_entry.name: dir_entry.stat()
                  for dir_entry in os.scandir(self.swap_dir)
              }
  
              data_files = {
                  swap_entry.data_filename: swap_entry
                  for swap_entry in self._swap_content.values()
              }
  
              # Get rid of stale files that are not referenced by any swap entry
              metadata_files = {
                  swap_entry.meta_filename
                  for swap_entry in self._swap_content.values()
              }
              metadata_files.add(self.TRACE_META_FILENAME)
              non_stale_files = data_files.keys() | metadata_files
              stale_files = stats.keys() - non_stale_files
              for filename in stale_files:
                  del stats[filename]
                  path = os.path.join(self.swap_dir, filename)
                  os.unlink(path)
  
              def by_mtime(path_stat):
                  path, stat = path_stat
                  return stat.st_mtime
  
              # Sort by modification time, so we discard the oldest caches
              total_size = 0
              discarded_swap_entries = set()
              for filename, stat in sorted(stats.items(), key=by_mtime):
                  total_size += stat.st_size
                  if total_size > self.max_swap_size:
                      try:
                          swap_entry = data_files[filename]
                      # That was not a data file
                      except KeyError:
                          continue
                      else:
                          discarded_swap_entries.add(swap_entry)
  
              # Update the swap content
              for swap_entry in discarded_swap_entries:
                  del self._swap_content[swap_entry.pd_desc_nf]
                  del stats[swap_entry.data_filename]
  
                  for filename in (swap_entry.meta_filename, swap_entry.data_filename):
                      path = os.path.join(self.swap_dir, filename)
                      os.unlink(path)
  
              self._swap_size = sum(
                  stats[swap_entry.data_filename].st_size
                  for swap_entry in self._swap_content.values()
              )
  
      def fetch(self, pd_desc, insert=True):
          """
          Fetch an entry from the cache or the swap.
  
          :param pd_desc: Descriptor to look for.
          :type pd_desc: PandasDataDesc
  
          :param insert: If ``True`` and if the fetch succeeds by loading the
              swap, the data is inserted in the cache.
          :type insert: bool
          """
          try:
              return self._cache[pd_desc]
          except KeyError as e:
              try:
                  path = self._swap_path_of(pd_desc)
              # If there is no swap, bail out
              except (ValueError, KeyError):
                  raise e
              else:
                  # Try to load the dataframe from that path
                  try:
                      if self.DATAFRAME_SWAP_FORMAT == 'parquet':
                          data = pd.read_parquet(path)
                      else:
                          raise ValueError('Dataframe swap format "{}" not handled'.format(self.DATAFRAME_SWAP_FORMAT))
                  except (OSError, pyarrow.lib.ArrowIOError):
                      raise e
                  else:
                      if insert:
                          # We have no idea of the cost of something coming from
                          # the cache
                          self.insert(pd_desc, data, write_swap=False, compute_cost=None)
  
                      return data
  
      def insert(self, pd_desc, data, compute_cost=None, write_swap=False, force_write_swap=False):
          """
          Insert an entry in the cache.
  
          :param pd_desc: Descriptor of the data to insert.
          :type pd_desc: PandasDataDesc
  
          :param data: Pandas data to insert.
          :type data: pandas.DataFrame or pandas.Series
  
          :param compute_cost: Time spent to compute the data in seconds.
          :type compute_cost: float or None
  
          :param write_swap: If ``True``, the data will be written to the swap as
              well so it can be quickly reloaded. Note that it will be subject to
              cost evaluation, so it might not result in anything actually
              written.
          :type write_swap: bool
  
          :param force_write_swap: If ``True``, bypass the computation vs swap
              cost comparison.
          :type force_write_swap: bool
          """
          self._cache[pd_desc] = data
          if compute_cost is not None:
              self._data_cost[pd_desc] = compute_cost
  
          if write_swap:
              self.write_swap(pd_desc, force=force_write_swap)
  
          self._scrub_mem()
  
      def _scrub_mem(self):
          if self.max_mem_size == math.inf:
              return
  
          mem_usage = sum(
              self._data_mem_usage(data)
              for data in self._cache.values()
          )
  
          if mem_usage > self.max_mem_size:
  
              # Make sure garbage collection occurred recently, to get the most
              # accurate refcount possible
              gc.collect()
              refcounts = {
                  pd_desc: sys.getrefcount(data)
                  for pd_desc, data in self._cache.items()
              }
              min_refcount = min(refcounts.values())
  
              # Low retention score means it's more likely to be evicted
              def retention_score(pd_desc_and_data):
                  pd_desc, data = pd_desc_and_data
  
                  # If we don't know the computation cost, assume it can be evicted cheaply
                  compute_cost = self._data_cost.get(pd_desc, 0)
  
                  if not compute_cost:
                      score = 0
                  else:
                      swap_cost = self._estimate_data_swap_cost(data)
                      # If it's already written back, make it cheaper to evict since
                      # the eviction itself is going to be cheap
                      if self._is_written_to_swap(pd_desc):
                          swap_cost /= 2
  
                      if swap_cost:
                          score = compute_cost / swap_cost
                      else:
                          score = 0
  
                  # Assume that more references to an object implies it will
                  # stay around for longer. Therefore, it's less interesting to
                  # remove it from this cache and pay the cost of reading/writing it to
                  # swap, since the memory will not be freed anyway.
                  #
                  # Normalize to the minimum refcount, so that the _cache and other
                  # structures where references are stored are discounted for sure.
                  return (refcounts[pd_desc] - min_refcount + 1) * score
  
              new_mem_usage = 0
              for pd_desc, data in sorted(self._cache.items(), key=retention_score):
                  new_mem_usage += self._data_mem_usage(data)
                  if new_mem_usage > self.max_mem_size:
                      self.evict(pd_desc)
  
      def evict(self, pd_desc):
          """
          Evict the given descriptor from memory.
  
          :param pd_desc: Descriptor to evict.
          :type pd_desc: PandasDataDesc
  
          If it would be cheaper to reload the data than to recompute them, they
          will be written to the swap area.
          """
          self.write_swap(pd_desc)
  
          try:
              del self._cache[pd_desc]
          except KeyError:
              pass
  
      def write_swap(self, pd_desc, force=False):
          """
          Write the given descriptor to the swap area if that would be faster to
          reload the data rather than recomputing it. If the descriptor is not in
          the cache or if there is no swap area, ignore it.
  
          :param pd_desc: Descriptor of the data to write to swap.
          :type pd_desc: PandasDataDesc
  
          :param force: If ``True``, bypass the compute vs swap cost comparison.
          :type force: bool
          """
          try:
              data = self._cache[pd_desc]
          except KeyError:
              pass
          else:
              if force or self._should_evict_to_swap(pd_desc, data):
                  self._write_swap(pd_desc, data)
  
      def write_swap_all(self):
          """
          Attempt to write all cached data to the swap.
          """
          for pd_desc in self._cache.keys():
              self.write_swap(pd_desc)
  
      def clear_event(self, event, raw=None):
          """
          Clear cache entries referencing a given event.
  
          :param event: Event to clear.
          :type event: str
  
          :param raw: If ``True``, only clear entries that refer to raw data. If
              ``False``, only clear entries that refer to non-raw data. If
              ``None``, ignore whether the descriptor is about raw data or not.
          :type raw: bool or None
          """
          self._cache = {
              pd_desc: data
              for pd_desc, data in self._cache.items()
              if not (
                  pd_desc.get('event') == event
                  and (
                      raw is None
                      or pd_desc.get('raw') == raw
                  )
              )
          }
  
      def clear_all_events(self, raw=None):
          """
          Same as :meth:`clear_event` but works on all events at once.
          """
          self._cache = {
              pd_desc: data
              for pd_desc, data in self._cache.items()
              if (
                  # Cache entries can be associated to something else than events
                  'event' not in pd_desc or
                  # Either we care about raw and we check, or blanket clear
                  raw is None or
                  pd_desc.get('raw') == raw
              )
          }
  
  
  class Trace(Loggable, TraceBase):
      """
      The Trace object is the LISA trace events parser.
  
      :param trace_path: File containing the trace
      :type trace_path: str
  
      :param events: events to be parsed. Since events can be loaded on-demand,
          that is optional but still recommended to improve trace parsing speed.
  
          .. seealso:: :meth:`df_event` for event formats accepted.
      :type events: list(str) or None
  
      :param strict_events: When ``True``, all the events specified in ``events``
          have to be present, and any other events will be assumed to not be
          present. This allows early failure and avoid the cost of lazy detection
          of events in very large traces.
      :type strict_events: bool
  
      :param plat_info: Platform info describing the target that this trace was
          collected on.
      :type plat_info: lisa.platforms.platinfo.PlatformInfo
  
      :param normalize_time: Make the first timestamp in the trace 0 instead
          of the system timestamp that was captured when tracing.
      :type normalize_time: bool
  
      :param parser: Optional trace parser to use as a backend. It must
          implement the API defined by :class:`TraceParserBase`, and will be
          called as ``parser(path=trace_path, events=events,
          needed_metadata={'time-range', ...})`` with the events that should be parsed.
          Other parameters must either have default values, or be pre-assigned
          using :func:`functools.partial`.
      :type parser: object or None
  
      :param plots_dir: directory where to save plots
      :type plots_dir: str
  
      :param sanitization_functions: Mapping of event name to sanitization
          function. Each function takes:
  
              * the trace instance
              * the name of the event
              * a dataframe of the raw event
              * a dictionary of aspects to sanitize
  
          These functions *must not* modify their input dataframe under any
          circumstances. They are required to make copies where appropriate.
      :type sanitization_functions: dict(str, collections.abc.Callable) or None
  
      :param max_mem_size: Maximum memory usage to be used for dataframe cache.
          Note that the peak memory usage can exceed that, as the cache can not
          forcefully evict an object from memory (it can only drop references to it).
          When ``None``, use illimited amount of memory.
      :type max_mem_size: int or None
  
      :param swap_dir: Swap directory used to store dataframes evicted from the
          cache. When ``None``, a hidden directory along the trace file is used.
      :type swap_dir: str or None
  
      :param enable_swap: If ``True``, the on-disk swap is enabled.
      :type enable_swap: bool
  
      :param max_swap_size: Maximum size of the swap directory. When ``None``,
          the max size is the size of the trace file.
      :type max_swap_size: int or None
  
      :param write_swap: Default value used for :meth:`df_event` ``write_swap``
          parameter.
      :type write_swap: bool
  
      :ivar start: The timestamp of the first trace event in the trace
      :ivar end: The timestamp of the last trace event in the trace
      :ivar time_range: Maximum timespan for all collected events
      :ivar window: Conveniency tuple of ``(start, end)``.
      :ivar available_events: Events available in the parsed trace, exposed as
          some kind of set-ish smart container. Querying for event might trigger
          the parsing of it.
  
      :Supporting more events:
          Subclasses of :class:`TraceParserBase` can usually auto-detect the
          event format, but there may be a number of reasons to pass a custom
          event parser:
  
              * The event format produced by a given kernel differs from the
                description bundled with the parser, leading to incorrect parse
                (missing field).
  
              * The event cannot be parsed in raw format in case text output of
                ``trace-cmd`` is used, because of a ``const char*`` field displayed
                as a pointer for example.
  
                .. seealso:: For events not following the regular field syntax,
                  use :class:`CustomFieldsTxtEventParser`
  
              * Automatic detection can take a heavy performance toll. This is
                why parsers needing descriptions will come with pre-defined
                descritption of most used events.
  
          Custom event parsers can be passed as extra parameters to the parser,
          which can be set manually::
  
              from functools import partial
  
              # Event parsers provided by TxtTraceParser can also be overridden
              # with user-provided ones in case they fail to parse what a given
              # kernel produced
              event_parsers = [
                  TxtEventParser('foobar', fields={'foo': int, 'bar': 'string'}, raw=True)
              ]
  
              # Pre-fill the "event_parsers" parameter of
              # TxtEventParser.from_dat() using a partial application.
              #
              # Note: you need to choose the parser appropriately for the
              # format of the trace, since the automatic filetype detection is
              # bypassed.
              parser = partial(TxtTraceParser.from_dat, event_parsers=event_parsers)
              trace = Trace('foobar.dat', parser=parser)
  
          .. warning:: Custom event parsers that are not types (such as
              :func:`functools.partial` values) will tie the on-disc swap entries
              to the parser :class:`Trace` instance, incurring higher
              :class:`pandas.DataFrame` load time when a new :class:`Trace`
              object is created.
      """
  
      def _select_userspace(source_event, meta_event, df):
          # tracing_mark_write is the name of the kernel function invoked when
          # writing to: /sys/kernel/debug/tracing/trace_marker
          df = df[df['func'] == 'tracing_mark_write']
          return (df, 'buf')
  
      def _select_trace_printk(source_event, meta_event, df):
          content_col = {
              'bprint': 'buf',
              'bputs': 'str',
          }[source_event]
  
          # Select on foobar function name with "trace_printk@func@foobar"
          func_prefix = 'func@'
          if meta_event.startswith(func_prefix):
              func_name = meta_event[len(func_prefix):]
              df = df[df['func'] == func_name].copy(deep=False)
              # Prepend the meta event name so it will be matched
              fake_event = meta_event.encode('ascii') + b': '
              df[content_col] = fake_event + df[content_col]
  
          return (df, content_col)
  
      _META_EVENT_SOURCE = {
          'userspace': {
              'print': _select_userspace,
          },
          'trace_printk': {
              'bprint': _select_trace_printk,
              'bputs': _select_trace_printk,
          },
      }
      """
      Define the source of each meta event.
  
      Meta events are events derived by parsing a field of another given real event.
  
      Meta events are event with a name formed as ``prefix:event``.
      This mapping has the format::
  
          prefix:
              source event: getter_func
              ... # Multiple sources are allowed
          ...
  
      ``getter_func`` takes the :class:`pandas.DataFrame` of the source event as
      input and returns a tuple of ``(dataframe, line_column)`` where:
  
          * ``dataframe`` is the source event dataframe, potentially filtered.
          * ``line_column`` is the name of the column containing the lines to
            parse.
      """
  
      def __init__(self,
          trace_path,
          plat_info=None,
          events=None,
          strict_events=False,
          normalize_time=False,
          parser=None,
          plots_dir=None,
          sanitization_functions=None,
  
          max_mem_size=None,
          swap_dir=None,
          enable_swap=True,
          max_swap_size=None,
          write_swap=True,
      ):
          super().__init__()
  
          sanitization_functions = sanitization_functions or {}
          self._sanitization_functions = {
              **self._SANITIZATION_FUNCTIONS,
              **sanitization_functions,
          }
  
          if enable_swap:
              if swap_dir is None:
                  basename = os.path.basename(trace_path)
                  swap_dir = os.path.join(
                      os.path.dirname(trace_path),
                      '.{}.lisa-swap'.format(basename)
                  )
                  try:
                      os.makedirs(swap_dir, exist_ok=True)
                  except OSError:
                      swap_dir = None
  
              if max_swap_size is None:
                  trace_size = os.stat(trace_path).st_size
                  max_swap_size = trace_size
          else:
              swap_dir = None
              max_swap_size = None
  
          self._cache = TraceCache.from_swap_dir(
              trace_path=trace_path,
              swap_dir=swap_dir,
              max_swap_size=max_swap_size,
              max_mem_size=max_mem_size,
          )
          # Initial scrub of the swap to discard unwanted data, honoring the
          # max_swap_size right from the beginning
          self._cache.scrub_swap()
          self._cache.to_swap_dir()
  
          self._write_swap = write_swap
          self.normalize_time = normalize_time
          self.trace_path = trace_path
  
          if parser is None:
              _, extension = os.path.splitext(trace_path)
              if extension == '.html':
                  parser = SysTraceParser.from_txt_file
              if extension == '.txt':
                  parser = TxtTraceParser.from_txt_file
              else:
                  parser = TxtTraceParser.from_dat
          self._parser = parser
  
          # The platform information used to run the experiments
          if plat_info is None:
              # Delay import to avoid circular dependency
              from lisa.platforms.platinfo import PlatformInfo
              plat_info = PlatformInfo()
          else:
              # Make a shallow copy so we can update it
              plat_info = copy.copy(plat_info)
  
          self._strict_events = strict_events
          self.available_events = _AvailableTraceEventsSet(self)
          self.plots_dir = plots_dir if plots_dir else os.path.dirname(trace_path)
  
          try:
              self._parseable_events = self._cache.get_metadata('parseable-events')
          except MissingMetadataError:
              self._parseable_events = {}
  
          if isinstance(events, str):
              raise ValueError('Events passed to Trace(events=...) must be a list of strings, not a string.')
  
          events = events if events is not None else []
          self.events = events
          # Pre-load the selected events
          if events:
              self._load_cache_raw_df(events, write_swap=True, allow_missing_events=not strict_events)
  
          # Register what we currently have
          self.plat_info = plat_info
          # Update the platform info with the data available from the trace once
          # the Trace is almost fully initialized
          self.plat_info = plat_info.add_trace_src(self)
  
  
      _CACHEABLE_METADATA = {
          'time-range',
          'cpus-count',
      }
      """
      Parser metadata that can safely be cached, i.e. that are serializable in
      the trace cache.
      """
  
      def get_metadata(self, key):
          """
          Get metadata from the underlying trace parser.
  
          .. seealso:: :meth:`TraceParserBase.get_metadata`
          """
          if key in self._CACHEABLE_METADATA:
              return self._get_cacheable_metadata(key)
          else:
              return self._get_metadata(key)
  
      def _get_metadata(self, key, parser=None):
          if parser is None:
              parser = self._get_parser(needed_metadata={key})
  
          return parser.get_metadata(key)
  
      def _get_cacheable_metadata(self, key, parser=None):
          try:
              value = self._cache.get_metadata(key)
          except MissingMetadataError:
              if parser is None:
                  value = self._get_metadata(key)
              else:
                  value = parser.get_metadata(key)
  
              self._cache.update_metadata({key: value})
  
          return value
  
      @classmethod
      def _is_meta_event(cls, event):
          sources = cls.get_event_sources(event)
          # If an event is not its own source, this is a meta-event by definition
          return sources[0] != event
  
      @classmethod
      def get_event_sources(cls, event):
          """
          Get the possible sources events of a given event.
  
          For normal events, this will just be a list with the event itself in
          it.
  
          For meta events, this will be the list of source events hosting that
          meta-event.
          """
          try:
              prefix, _ = event.split('@', 1)
          except ValueError:
              return [event]
  
          try:
              return sorted(cls._META_EVENT_SOURCE[prefix].keys())
          except KeyError:
              return [event]
  
      @property
      # Memoization is necessary to ensure the parser always gets the same name
      # on a given Trace instance when the parser is not a type.
      @lru_memoized(first_param_maxsize=None, other_params_maxsize=None)
      def trace_state(self):
          parser = self._parser
          # The parser type will potentially change the exact content in raw
          # dataframes
          def get_name(parser):
              return '{}.{}'.format(
                  parser.__module__,
                  parser.__qualname__
              )
  
          try:
              parser_name = get_name(parser)
          # If the parser is an instance of something, we cannot safely track its
          # state so just make a unique name for it
          except AttributeError:
              parser_name = '{}-instance:{}'.format(
                  get_name(parser.__class__),
                  uuid.uuid4().hex
              )
  
          return (self.normalize_time, parser_name)
  
      @property
      @memoized
      def cpus_count(self):
          try:
              return self.plat_info['cpus-count']
          except KeyError:
              try:
                  self.get_metadata('cpus-count')
              # If we don't know the number of CPUs, check the trace for the
              # highest-numbered CPU that traced an event.
              except MissingMetadataError:
                  # Sched_switch should be enough if it's available
                  if 'sched_switch' in self.available_events:
                      checked_events = ['sched_switch']
                  # This is pretty costly, as it will trigger parsing of all
                  # events in the trace, so only do it as last resort
                  else:
                      checked_events = self.available_events
  
                  max_cpu = max(
                      int(self.df_event(e)['__cpu'].max())
                      for e in checked_events
                  )
                  count = max_cpu + 1
                  self.get_logger().debug("Estimated CPU count from trace: {}".format(count))
  
              return count
  
      @classmethod
      @contextlib.contextmanager
      def from_target(cls, target, events=None, buffer_size=10240, filepath=None, **kwargs):
          """
          Context manager that can be used to collect a :class:`Trace` directly
          from a :class:`lisa.target.Target` without needing to setup an
          :class:`FtraceCollector`.
  
          **Example**::
  
              from lisa.trace import Trace
              from lisa.target import Target
  
              target = Target.from_default_conf()
  
              with Trace.from_target(target, events=['sched_switch', 'sched_wakeup']) as trace:
                  target.execute('echo hello world')
                  # DO NOT USE trace object inside the `with` statement
  
              trace.analysis.tasks.plot_tasks_total_residency(filepath='plot.png')
  
  
          :param target: Target to connect to.
          :type target: Target
  
          :param events: ftrace events to collect and parse in the trace.
          :type events: list(str)
  
          :param buffer_size: Size of the ftrace ring buffer.
          :type buffer_size: int
  
          :param filepath: If set, the trace file will be saved at that location.
              Otherwise, a temporary file is created and removed as soon as the
              parsing is finished.
          :type filepath: str or None
  
          :Variable keyword arguments: Forwarded to :class:`Trace`.
          """
          ftrace_coll = FtraceCollector(target, events=events, buffer_size=buffer_size)
          plat_info = target.plat_info
  
          class TraceProxy(TraceBase):
              def get_view(self, *args, **kwargs):
                  return self.base_trace.get_view(*args, **kwargs)
  
              def __getattr__(self, attr):
                  try:
                      base_trace = self.__dict__['base_trace']
                  except KeyError:
                      raise RuntimeError('The trace instance can only be used outside its "with" statement.')
                  else:
                      return getattr(base_trace, attr)
  
          proxy = TraceProxy()
  
          with ftrace_coll:
              yield proxy
  
          if filepath:
              cm = nullcontext(filepath)
          else:
              @contextlib.contextmanager
              def cm_func():
                  with tempfile.NamedTemporaryFile(suffix='.dat', delete=True) as temp:
                      yield temp.name
  
              cm = cm_func()
  
          with cm as path:
              ftrace_coll.get_data(path)
              trace = cls(
                  path,
                  events=events,
                  strict_events=True,
                  plat_info=plat_info,
                  # Disable swap if the folder is going to disappear
                  enable_swap=True if filepath else False,
                  **kwargs
              )
  
          proxy.base_trace = trace
  
      def _get_parser(self, events=tuple(), needed_metadata=None, update_metadata=True):
          logger = self.get_logger()
          path = self.trace_path
          events = set(events)
          needed_metadata = set(needed_metadata or [])
          parser = self._parser(path=path, events=events, needed_metadata=needed_metadata)
  
          # While we are at it, gather a bunch of metadata. Since we did not
          # explicitly asked for it, the parser will only give
          # it if it was a cheap byproduct.
          if update_metadata:
  
              # Since we got a parser here, use it to get basetime/endtime as well
              with contextlib.suppress(MissingMetadataError):
                  self._get_time_range(parser=parser)
  
              # Populate the list of available events, and inform the rest of the
              # code that this list is definitive.
              try:
                  available_events = self._get_metadata('available-events', parser=parser)
              except MissingMetadataError:
                  pass
              else:
                  self._update_parseable_events({
                      event: True
                      for event in available_events
                  })
                  self._strict_events = True
  
          return parser
  
      def _update_parseable_events(self, mapping):
          self._parseable_events.update(mapping)
          self._cache.update_metadata({
              'parseable-events': self._parseable_events,
          })
          return self._parseable_events
  
      @property
      @memoized
      def basetime(self):
          """
          First absolute timestamp available in the trace.
          """
          return self._get_time_range()[0]
  
      @property
      @memoized
      def endtime(self):
          """
          Timestamp of when the tracing stopped.
  
          .. note:: With some parsers, that might be the timestamp of the last
              recorded event instead if the trace end timestamp was not recorded.
          """
          return self._get_time_range()[1]
  
      def _get_time_range(self, parser=None):
          return self._get_cacheable_metadata('time-range', parser)
  
      def df_event(self, event, raw=None, window=None, signals=None, signals_init=True, compress_signals_init=False, write_swap=None):
          """
          Get a dataframe containing all occurrences of the specified trace event
          in the parsed trace.
  
          :param event: Trace event name.
  
              In addition to actual events, the following formats for meta events
              are supported:
  
              * ``trace_printk@``: the event will be assumed to be embedded in
                textual form inside the field of another event as a string,
                typically emitted using the ``trace_printk()`` kernel function:
  
                .. code-block:: C
  
                    // trace.df_event('trace_printk@event')
                    void foo(void) {
                        trace_printk("event: optional_positional_field field1=foo field2=42");
                    }
  
              * ``trace_printk@func@``: the event name will be the name of the
                function calling trace_printk:
  
                .. code-block:: C
  
                    // trace.df_event('trace_printk@func@foo')
                    void foo(void) {
                        trace_printk("optional_positional_field field1=foo field2=42")
                    }
  
              * ``userspace@``: the event is generated by userspace:
  
                .. code-block:: shell
  
                    # trace.df_event('userspace@event')
                    echo "event: optional_positional_field field1=foo field2=42" > /sys/kernel/debug/tracing/trace_marker
  
              .. note:: All meta event names are expected to be valid C language
                  identifiers. Usage of other characters will prevent correct
                  parsing.
  
          :type event: str
  
          :param window: Return a dataframe sliced to fit the given window (in
              seconds). Note that ``signals_init=True`` will result in including
              more rows than what you might expect.
          :type window: tuple(float, float)
  
          :param signals: List of signals to fixup if ``signals_init == True``.
              If left to ``None``, :meth:`lisa.datautils.SignalDesc.from_event`
              will be used to infer a list of default signals.
          :type signals: list(SignalDesc)
  
          :param signals_init: If ``True``, an initial value is provided for each
              signal that is multiplexed in that dataframe.
  
              .. seealso::
  
                  :class:`lisa.datautils.SignalDesc` and
                  :func:`lisa.datautils.df_window_signals`.
          :type signals_init: bool
  
          :param compress_signals_init: Give a timestamp very close to the
              beginning of the sliced dataframe to rows that are added by
              ``signals_init``. This allows keeping a very close time span
              without introducing duplicate indices.
          :type compress_signals_init: bool
  
          :param write_swap: If ``True``, the dataframe will be written to the
              swap area when meeting the following conditions:
  
                  * This trace has a swap directory
                  * Computing the dataframe takes more time than the estimated
                    time it takes to write it to the cache.
          :type write_swap: bool
          """
  
          sanitization_f = self._sanitization_functions.get(event)
  
          # Make sure no `None` value flies around in the cache, since it's
          # not uniquely identifying a dataframe
          orig_raw = raw
          if raw is None:
              if sanitization_f:
                  raw = False
              else:
                  raw = True
  
          if raw:
              sanitization_f = None
          elif orig_raw == False and not sanitization_f:
              raise ValueError('Sanitized dataframe for {} does not exist, please pass raw=True or raw=None'.format(event))
  
          if raw:
              # Make sure all raw descriptors are made the same way, to avoid
              # missed sharing opportunities
              spec = self._make_raw_pd_desc_spec(event)
          else:
              spec = dict(
                  event=event,
                  raw=raw,
                  trace_state=self.trace_state,
              )
  
          if window is not None:
              signals = signals if signals else SignalDesc.from_event(event)
              cols_list = [
                  signal_desc.fields
                  for signal_desc in signals
              ]
              spec.update(
                  window=window,
                  signals=cols_list,
                  signals_init=signals_init,
                  compress_signals_init=compress_signals_init,
              )
  
          if not raw:
              spec.update(
                  rename_cols=True,
                  sanitization=sanitization_f.__qualname__ if sanitization_f else None,
              )
  
          pd_desc = PandasDataDesc(spec=spec)
  
          try:
              df = self._cache.fetch(pd_desc, insert=True)
          except KeyError:
              df = self._load_df(pd_desc, sanitization_f=sanitization_f, write_swap=write_swap)
  
          if df.empty:
              raise MissingTraceEventError(
                  TraceEventChecker(event),
                  available_events=self.available_events,
              )
  
          df.name = event
          return df
  
      def _make_raw_pd_desc(self, event):
          spec = self._make_raw_pd_desc_spec(event)
          return PandasDataDesc(spec=spec)
  
      def _make_raw_pd_desc_spec(self, event):
          return dict(
              event=event,
              raw=True,
              trace_state=self.trace_state,
          )
  
      def _load_df(self, pd_desc, sanitization_f=None, write_swap=None):
          raw = pd_desc['raw']
          event = pd_desc['event']
  
          # Do not even bother loading the event if we know it cannot be
          # there. This avoids some OSError in case the trace file has
          # disappeared
          if self._strict_events and not self._is_meta_event(event) and event not in self.available_events:
              raise MissingTraceEventError(event, available_events=self.available_events)
  
          if write_swap is None:
              write_swap = self._write_swap
  
          df = self._load_cache_raw_df([event], write_swap=True)[event]
  
          if sanitization_f:
              # Evict the raw dataframe once we got the sanitized version, since
              # we are unlikely to reuse it again
              self._cache.evict(self._make_raw_pd_desc(event))
  
              # We can ask to sanitize various aspects of the dataframe.
              # Adding a new aspect can be done without modifying existing
              # sanitization functions, as long as the default is the
              # previous behavior
              aspects = dict(
                  rename_cols=pd_desc['rename_cols'],
              )
              with measure_time() as measure:
                  df = sanitization_f(self, event, df, aspects=aspects)
              sanitization_time = measure.exclusive_delta
          else:
              sanitization_time = 0
  
          window = pd_desc.get('window')
          if window is not None:
              signals_init = pd_desc['signals_init']
              compress_signals_init = pd_desc['compress_signals_init']
              cols_list = pd_desc['signals']
              signals = [SignalDesc(event, cols) for cols in cols_list]
  
              with measure_time() as measure:
                  if signals_init and signals:
                      df = df_window_signals(df, window, signals, compress_init=compress_signals_init)
                  else:
                      df = df_window(df, window, method='pre')
  
              windowing_time = measure.exclusive_delta
          else:
              windowing_time = 0
  
          compute_cost = sanitization_time + windowing_time
          self._cache.insert(pd_desc, df, compute_cost=compute_cost, write_swap=write_swap)
          return df
  
      def _load_cache_raw_df(self, events, write_swap, allow_missing_events=False):
          insert_kwargs = dict(
              write_swap=write_swap,
              # For raw dataframe, always write in the swap area if asked for
              # since parsing cost is known to be high
              force_write_swap=True,
          )
  
          # Get the raw dataframe from the cache if possible
          def try_from_cache(event):
              pd_desc = self._make_raw_pd_desc(event)
              try:
                  # The caller is responsible of inserting in the cache if
                  # necessary
                  df = self._cache.fetch(pd_desc, insert=False)
              except KeyError:
                  return None
              else:
                  self._cache.insert(pd_desc, df, **insert_kwargs)
                  return df
  
          from_cache = {
              event: try_from_cache(event)
              for event in events
          }
  
          from_cache = {
              event: df
              for event, df in from_cache.items()
              if df is not None
          }
  
          # Load the remaining events from the trace directly
          events_to_load = sorted(set(events) - from_cache.keys())
          from_trace = self._load_raw_df(events_to_load)
  
          for event, df in from_trace.items():
              pd_desc = self._make_raw_pd_desc(event)
              self._cache.insert(pd_desc, df, **insert_kwargs)
  
          df_map = {**from_cache, **from_trace}
          missing_events = set(events) - df_map.keys()
          if missing_events:
              if allow_missing_events:
                  self.get_logger().warning('Events {} not found in the trace: {}'.format(
                      ', '.join(sorted(missing_events)),
                      self.trace_path,
                  ))
              else:
                  raise MissingTraceEventError(missing_events)
  
          return df_map
  
      def _apply_normalize_time(self, df, inplace):
          df = df if inplace else df.copy(deep=False)
  
          if self.normalize_time:
              df.index -= self.basetime
  
          return df
  
      def _mp_parse_worker(self, event):
          # Do not update the metadata to avoid concurrency issues while updating
          # the cache
          parser = self._get_parser([event], update_metadata=False)
  
          try:
              data = parser.parse_event(event)
          except MissingTraceEventError as e:
              data = e
          else:
              data = self._apply_normalize_time(data, inplace=True)
  
          return data
  
      def _parse_raw_events(self, events):
          if not events:
              return {}
  
          nr_processes = min(
              len(events),
              multiprocessing.cpu_count(),
          )
          chunk_size = int(math.ceil(len(events) / nr_processes))
  
          # Only use multiprocessing if there is no memory limit, since the peak
          # consumption will increase
          use_mp = self._cache.max_mem_size >= math.inf and nr_processes > 1
  
          if use_mp:
              with multiprocessing.Pool(processes=nr_processes) as pool:
                  data_list = pool.map(self._mp_parse_worker, events, chunksize=chunk_size)
  
              df_map = {
                  event: df
                  for event, df in zip(
                      events,
                      data_list,
                  )
                  # similar to best_effort=True
                  if not isinstance(df, BaseException)
              }
          else:
              parser = self._get_parser(events, update_metadata=True)
              df_map = parser.parse_events(events, best_effort=True)
  
              for df in df_map.values():
                  self._apply_normalize_time(df, inplace=True)
  
          return df_map
  
      def _parse_meta_events(self, meta_events):
          if not meta_events:
              return {}
  
          # Gather the infor to parse the meta event
          def make_spec(meta_event):
              prefix, event = meta_event.split('@', 1)
              data_getters = self._META_EVENT_SOURCE[prefix]
              return (meta_event, event, data_getters)
  
          meta_specs = list(map(make_spec, meta_events))
  
          # Map each trimmed event name back to its meta event name
          events_map = {
              event: meta_event
              for meta_event, event, _ in meta_specs
          }
  
          # Explode per source event
          meta_specs = [
              (meta_event, event, source_event, source_getter)
              for (meta_event, event, data_getters) in meta_specs
              for source_event, source_getter in data_getters.items()
          ]
  
          def get_missing(df_map):
              return events_map.keys() - df_map.keys()
  
          df_map = {}
          # Group all the meta events by their source event, so we process source
          # dataframes one by one
          for source_event, specs in groupby(meta_specs, key=itemgetter(2)):
              try:
                  df = self.df_event(source_event)
              except MissingTraceEventError:
                  pass
              else:
                  # Add all the header fields from the source dataframes
                  extra_fields = [x for x in df.columns if x.startswith('__')]
                  merged_df = df[extra_fields]
  
                  for (meta_event, event, source_event, source_getter) in specs:
                      source_df, line_field = source_getter(source_event, event, df)
                      try:
                          parser = MetaTxtTraceParser(
                              lines=source_df[line_field],
                              time=source_df.index,
                              merged_df=merged_df,
                              events=[event],
                          )
                      # An empty dataframe would trigger an exception from
                      # MetaTxtTraceParser, so we just skip over it and if the
                      # event cannot be found anywhere, we will raise an
                      # exception.
                      #
                      # Also, some android kernels hide trace_printk() strings
                      # which prevents from parsing anything at all ...
                      except ValueError:
                          continue
                      else:
                          try:
                              meta_event_df = parser.parse_event(event)
                          except MissingTraceEventError:
                              continue
                              # In case a meta-event is spread among multiple
                              # events, we get all the dataframes and concatenate
                              # them together
                          else:
                              df_map.setdefault(event, []).append(meta_event_df)
  
                      if not get_missing(df_map):
                          break
  
          def concat(df_list):
              if len(df_list) > 1:
                  # As of pandas == 1.0.3, concatenating dataframe with nullable
                  # columns will give an object dtype and both NaN and <NA> if
                  # that column does not exist in all dataframes. This is quite
                  # annoying but there is no straightforward way of working
                  # around that.
                  df = pd.concat(df_list, copy=False)
                  df.sort_index(inplace=True)
                  return df
              # Avoid creating a new dataframe and sorting the index
              # unnecessarily if there is only one
              else:
                  return df_list[0]
  
          df_map = {
              event: concat(df_list)
              for event, df_list in df_map.items()
          }
  
          # On some parsers, meta events are treated as regular events so attempt
          # to load them from there as well
          for event in get_missing(df_map):
              with contextlib.suppress(MissingTraceEventError):
                  df_map[event] = self.df_event(event, raw=True)
  
          return {
              events_map[event]: df
              for event, df in df_map.items()
          }
  
      def _load_raw_df(self, events):
          events = set(events)
          if not events:
              return {}
  
          meta_events = set(filter(self._is_meta_event, events))
          regular_events = events - meta_events
  
          df_map = {
              **self._parse_raw_events(regular_events),
              **self._parse_meta_events(meta_events),
          }
  
          # Save some memory by changing values of this column into an category
          categorical_fields = [
              '__comm',
              'comm',
          ]
          for df in df_map.values():
              for field in categorical_fields:
                  with contextlib.suppress(KeyError):
                      df[field] = df[field].astype('category', copy=False)
  
          # remember the events that we tried to parse and that turned out to not be available
          self._update_parseable_events({
              event: (event in df_map)
              for event in events
          })
  
          return df_map
  
      @memoized
      def _get_task_maps(self):
          """
          Give the mapping from PID to task names, and the opposite.
  
          The names or PIDs are listed in appearance order.
          """
  
  
          # Keep only the values, in appearance order according to the timestamp
          # index
          def finalize(df, key_col, value_col, key_type, value_type):
              # Aggregate the values for each key and convert to python types
              mapping = {}
              grouped = df.groupby([key_col], observed=True, sort=False)
              for key, subdf in grouped:
                  values = subdf[value_col].apply(value_type).to_list()
                  key = key_type(key)
                  mapping[key] = values
  
              return mapping
  
          mapping_df_list = []
          def _load(event, name_col, pid_col):
              df = self.df_event(event)
  
              # Get a Time column
              df = df.reset_index()
              grouped = df.groupby([name_col, pid_col], observed=True, sort=False)
  
              # Get timestamp of first occurrences of each key/value combinations
              mapping_df = grouped.head(1)
              mapping_df = mapping_df[['Time', name_col, pid_col]]
              mapping_df.rename({name_col: 'name', pid_col: 'pid'}, axis=1, inplace=True)
              mapping_df_list.append(mapping_df)
  
          def load(event, *args, **kwargs):
              # All events have a __comm and __pid columns, so use it as well
              _load(event, '__comm', '__pid')
              _load(event, *args, **kwargs)
  
          # Import here to avoid circular dependency
          from lisa.analysis.load_tracking import LoadTrackingAnalysis
          # All events with a "comm" and "pid" column
          events = {
              'sched_wakeup',
              'sched_wakeup_new',
              *LoadTrackingAnalysis._SCHED_PELT_SE_NAMES,
          }
          for event in events:
              # Test each event independently, to make sure they will be parsed
              # if necessary
              if event in self.available_events:
                  load(event, 'comm', 'pid')
  
          if 'sched_switch' in self.available_events:
              load('sched_switch', 'prev_comm', 'prev_pid')
              load('sched_switch', 'next_comm', 'next_pid')
  
          if not mapping_df_list:
              raise MissingTraceEventError(sorted(events) + ['sched_switch'], available_events=self.available_events)
  
          df = pd.concat(mapping_df_list)
          # Sort by order of appearance
          df.sort_values(by=['Time'], inplace=True)
          # Remove duplicated name/pid mapping and only keep the first appearance
          df = df_deduplicate(df, consecutives=False, keep='first', cols=['name', 'pid'])
  
          forbidden_names = {
              # <idle> is invented by trace-cmd, no event field contain this
              # value, so it's useless (and actually harmful, since it will
              # introduce a task that cannot be found in that trace)
              '<idle>',
              # This name appears when trace-cmd could not resolve the task name.
              # Ignore it since it's not a valid name, and we probably managed
              # to resolve it by looking at more events anyway.
              '<...>',
              # sched entity PELT events for task groups will get a comm="(null)"
              '(null)',
          }
          df = df[~df['name'].isin(forbidden_names)]
  
          name_to_pid = finalize(df, 'name', 'pid', str, int)
          pid_to_name = finalize(df, 'pid', 'name', int, str)
  
          return (name_to_pid, pid_to_name)
  
      @property
      def _task_name_map(self):
          return self._get_task_maps()[0]
  
      @property
      def _task_pid_map(self):
          return self._get_task_maps()[1]
  
      def has_events(self, events):
          """
          Returns True if the specified event is present in the parsed trace,
          False otherwise.
  
          :param events: trace event name or list of trace events
          :type events: str or list(str) or TraceEventCheckerBase
          """
          if isinstance(events, str):
              return events in self.available_events
          elif isinstance(events, TraceEventCheckerBase):
              try:
                  events.check_events(self.available_events)
              except MissingTraceEventError:
                  return False
              else:
                  return True
          else:
              # Check each event independently in case they have not been parsed
              # yet.
              return all(
                  event in self.available_events
                  for event in events
              )
  
      def get_view(self, window, **kwargs):
          return TraceView(self, window, **kwargs)
  
      @property
      def start(self):
          return 0 if self.normalize_time else self.basetime
  
      @property
      def end(self):
          time_range = self.endtime - self.basetime
          return self.start + time_range
  
      def get_task_name_pids(self, name, ignore_fork=True):
          """
          Get the PIDs of all tasks with the specified name.
  
          The same PID can have different task names, mainly because once a task
          is generated it inherits the parent name and then its name is updated
          to represent what the task really is.
  
          :param name: task name
          :type name: str
  
          :param ignore_fork: Hide the PIDs of tasks that initially had ``name``
              but were later renamed. This is common for shell processes for
              example, which fork a new task, inheriting the shell name, and then
              being renamed with the final "real" task name
          :type ignore_fork: bool
  
          :return: a list of PID for tasks which name matches the required one.
          """
          pids = self._task_name_map[name]
  
          if ignore_fork:
              pids = [
                  pid
                  for pid in pids
                  # Only keep the PID if its last name was the name we are
                  # looking for.
                  if self._task_pid_map[pid][-1] == name
              ]
  
          return pids
  
      @deprecate('This method has been deprecated and is an alias',
          deprecated_in='2.0',
          removed_in='2.1',
          replaced_by=get_task_name_pids,
      )
      def get_task_by_name(self, name):
          return self.get_task_name_pids(name, ignore_fork=True)
  
      def get_task_pid_names(self, pid):
          """
          Get the all the names of the task(s) with the specified PID, in
          appearance order.
  
          The same PID can have different task names, mainly because once a task
          is generated it inherits the parent name and then its name is
          updated to represent what the task really is.
  
          :param name: task PID
          :type name: int
  
          :return: the name of the task which PID matches the required one,
                   the last time they ran in the current trace
          """
          return self._task_pid_map[pid]
  
      @deprecate('This function raises exceptions when faced with ambiguity instead of giving the choice to the user',
          deprecated_in='2.0',
          removed_in='2.1',
          replaced_by=get_task_pid_names,
      )
      def get_task_by_pid(self, pid):
          """
          Get the name of the task with the specified PID.
  
          The same PID can have different task names, mainly because once a task
          is generated it inherits the parent name and then its name is
          updated to represent what the task really is.
  
          This API works under the assumption that a task name is updated at
          most one time and it always report the name the task had the last time
          it has been scheduled for execution in the current trace.
  
          :param name: task PID
          :type name: int
  
          :return: the name of the task which PID matches the required one,
                   the last time they ran in the current trace
          """
          name_list = self.get_task_pid_names(pid)
  
          if len(name_list) > 2:
              raise RuntimeError('The PID {} had more than two names in its life: {}'.format(
                  pid, name_list,
              ))
  
          return name_list[-1]
  
      def get_task_ids(self, task, update=True):
          """
          Similar to :meth:`get_task_id` but returns a list with all the
          combinations, instead of raising an exception.
  
          :param task: Either the task name, the task PID, or a tuple ``(pid, comm)``
          :type task: int or str or tuple(int, str)
  
          :param update: If a partially-filled :class:`TaskID` is passed (one of
              the fields set to ``None``), returns a complete :class:`TaskID`
              instead of leaving the ``None`` fields.
          :type update: bool
          """
  
          def comm_to_pid(comm):
              try:
                  pid_list = self._task_name_map[comm]
              except IndexError:
                  raise ValueError('trace does not have any task named "{}"'.format(comm))
  
              return pid_list
  
          def pid_to_comm(pid):
              try:
                  comm_list = self._task_pid_map[pid]
              except IndexError:
                  raise ValueError('trace does not have any task PID {}'.format(pid))
  
              return comm_list
  
          if isinstance(task, str):
              task_ids = [
                  TaskID(pid=pid, comm=task)
                  for pid in comm_to_pid(task)
              ]
          elif isinstance(task, Number):
              task_ids = [
                  TaskID(pid=task, comm=comm)
                  for comm in pid_to_comm(task)
              ]
          else:
              pid, comm = task
              if pid is None and comm is None:
                  raise ValueError('TaskID needs to have at least one of PID or comm specified')
  
              if update and (pid is None or comm is None):
                  non_none = pid if comm is None else comm
                  task_ids = self.get_task_ids(non_none)
              else:
                  task_ids = [TaskID(pid=pid, comm=comm)]
  
          return task_ids
  
      def get_task_id(self, task, update=True):
          """
          Helper that resolves a task PID or name to a :class:`TaskID`.
  
          :param task: Either the task name, the task PID, or a tuple ``(pid, comm)``
          :type task: int or str or tuple(int, str)
  
          :param update: If a partially-filled :class:`TaskID` is passed (one of
              the fields set to ``None``), returns a complete :class:`TaskID`
              instead of leaving the ``None`` fields.
          :type update: bool
  
          :raises ValueError: If there the input matches multiple tasks in the trace.
              See :meth:`get_task_ids` to get all the ambiguous alternatives
              instead of an exception.
          """
          task_ids = self.get_task_ids(task, update=update)
          if len(task_ids) > 1:
              raise ValueError('More than one TaskID matching: {}'.format(task_ids))
  
          return task_ids[0]
  
      @deprecate(deprecated_in='2.0', removed_in='2.1', replaced_by=get_task_id)
      def get_task_pid(self, task):
          """
          Helper that takes either a name or a PID and always returns a PID
  
          :param task: Either the task name or the task PID
          :type task: int or str or tuple(int, str)
          """
          return self.get_task_id(task).pid
  
      def get_tasks(self):
          """
          Get a dictionary of all the tasks in the Trace.
  
          :return: a dictionary which maps each PID to the corresponding list of
                   task name
          """
          return self._task_pid_map
  
      @property
      @memoized
      def task_ids(self):
          """
          List of all the :class:`TaskID` in the trace, sorted by PID.
          """
          return [
              TaskID(pid=pid, comm=comm)
              for pid, comms in sorted(self._task_pid_map.items(), key=itemgetter(0))
              for comm in comms
          ]
  
      def show(self):
          """
          Open the parsed trace using the most appropriate native viewer.
  
          The native viewer depends on the specified trace format:
          - ftrace: open using kernelshark
          - systrace: open using a browser
  
          In both cases the native viewer is assumed to be available in the host
          machine.
          """
          path = self.trace_path
          if path.endswith('.dat'):
              cmd = 'kernelshark'
          else:
              cmd = 'xdg-open'
  
          return os.popen("{} {}".format(
              cmd,
              shlex.quote(path)
          ))
  
  ###############################################################################
  # Trace Events Sanitize Methods
  ###############################################################################
  
      _SANITIZATION_FUNCTIONS = {}
      def _sanitize_event(event, mapping=_SANITIZATION_FUNCTIONS):
          """
          Sanitization functions must not modify their input.
          """
          def decorator(f):
              mapping[event] = f
              return f
          return decorator
  
      @_sanitize_event('sched_switch')
      def _sanitize_sched_switch(self, event, df, aspects):
          """
          If ``prev_state`` is a string, turn it back into an integer state by
          parsing it.
          """
          if df['prev_state'].dtype.name == 'string':
              # Avoid circular dependency issue by importing at the last moment
              from lisa.analysis.tasks import TaskState
              df = df.copy(deep=False)
              df['prev_state'] = df['prev_state'].apply(TaskState.from_sched_switch_str).astype('uint16', copy=False)
  
          # Save a lot of memory by using category for strings
          for col in ('next_comm', 'prev_comm'):
              df[col] = df[col].astype('category', copy=False)
  
          return df
  
      @_sanitize_event('thermal_power_cpu_limit')
      @_sanitize_event('thermal_power_cpu_get_power')
      def _sanitize_thermal_power_cpu(self, event, df, aspects):
  
          def f(mask):
              # Replace '00000000,0000000f' format in more usable int
              return int(mask.replace(',', ''), 16)
  
          df = df.copy(deep=False)
          df['cpus'] = df['cpus'].apply(f)
          return df
  
      @_sanitize_event('print')
      @_sanitize_event('bprint')
      @_sanitize_event('bputs')
      def _sanitize_print(self, event, df, aspects):
          df = df.copy(deep=False)
          # Reduce memory usage and speedup selection based on function
          with contextlib.suppress(KeyError):
              df['func'] = df['func'].astype('category', copy=False)
          return df
  
      @staticmethod
      def _expand_bitmask_field(mask):
          """
          Turn a bitmask (like cpu_mask) formated by trace-cmd in non-raw mode
          into a list of integers for each bitmask position that is set.
  
          ``mask`` is a string with comma-separated hex numbers like
          "000001,12345,..."
          """
          numbers = mask.split(',')
  
          # hex number, so 4 bit per digit
          nr_bits = len(numbers[0]) * 4
  
          def bit_pos(number):
              # Little endian
              number = int(number, base=16)
              return [
                  i
                  for i in range(nr_bits)
                  if number & (1 << i)
              ]
  
          return [
              i + (nr_bits * offset)
              for offset, positions in enumerate(
                  # LSB is in the number at the end of the list so we reverse it
                  map(bit_pos, reversed(numbers))
              )
              for i in positions
          ]
  
      @_sanitize_event('ipi_raise')
      def _sanitize_ipi_raise(self, event, df, aspects):
          df = df.copy(deep=False)
          df['target_cpus'] = df['target_cpus'].apply(self._expand_bitmask_field)
          df['reason'] = df['reason'].str.strip('()')
          return df
  
      @_sanitize_event('ipi_entry')
      @_sanitize_event('ipi_exit')
      def _sanitize_ipi_enty_exit(self, event, df, aspects):
          df = df.copy(deep=False)
          df['reason'] = df['reason'].str.strip('()')
          return df
  
  class TraceEventCheckerBase(abc.ABC, Loggable):
      """
      ABC for events checker classes.
  
      Event checking can be achieved using a boolean expression on expected
      events.
  
      :param check: Check that the listed events are present in the
          ``self.trace`` attribute of the instance on which the decorated
          method is applied.  If no such attribute is found, no check will be
          done.
      :type check: bool
      """
  
      def __init__(self, check=True):
          self.check = check
  
      @abc.abstractmethod
      def check_events(self, event_set):
          """
          Check that certain trace events are available in the given set of
          events.
  
          :raises: MissingTraceEventError if some events are not available
          """
          pass
  
      @abc.abstractmethod
      def get_all_events(self):
          """
          Return a set of all events that are checked by this checker.
  
          That may be a superset of events that are strictly required, when the
          checker checks a logical OR combination of events for example.
          """
          pass
  
      def __call__(self, f):
          """
          Decorator for methods that require some given trace events
  
          :param events: The list of required events
          :type events: list(str or TraceEventCheckerBase)
  
          The decorated method must operate on instances that have a ``self.trace``
          attribute.
  
          If some event requirements have already been defined for it (it has a
          `used_events` attribute, i.e. it has already been decorated), these
          will be combined with the new requirements using an
          :class`AndTraceEventChecker`.
          """
          def unwrap_down_to(obj):
              return hasattr(obj, 'used_events')
  
          try:
              # we want to see through all other kinds of wrappers, down to the
              # one that matters to us
              unwrapped_f = inspect.unwrap(f, stop=unwrap_down_to)
              used_events = unwrapped_f.used_events
          except AttributeError:
              checker = self
          else:
              # Update the existing checker inplace to avoid adding an extra
              # level of wrappers.
              checker = AndTraceEventChecker([self, used_events])
              unwrapped_f.used_events = checker
              return f
  
          sig = inspect.signature(f)
          if self.check and sig.parameters:
              @wraps(f)
              def wrapper(self, *args, **kwargs):
                  try:
                      trace = self.trace
                  # If there is no "trace" attribute, silently skip the check. This
                  # allows using the decorator for documentation and chaining purpose
                  # without having an actual trace to work on.
                  except AttributeError:
                      pass
                  else:
                      checker.check_events(trace.available_events)
  
                  return f(self, *args, **kwargs)
  
          # If the decorated object takes no parameters, we cannot check anything
          else:
              @wraps(f)
              def wrapper(*args, **kwargs):
                  return f(*args, **kwargs)
  
          # Set an attribute on the wrapper itself, so it can be e.g. added
          # to the method documentation
          wrapper.used_events = checker
          return wrapper
  
      def __and__(self, other):
          """
          Combine two event checkers into one that checks the presence of both.
  
          .. seealso:: :class:`AndTraceEventChecker`
          """
          return AndTraceEventChecker([self, other])
  
      def __or__(self, other):
          """
          Combine two event checkers into one that checks the presence of either
          of them.
  
          .. seealso:: :class:`OrTraceEventChecker`
          """
          return OrTraceEventChecker([self, other])
  
      def __matmul__(self, other):
          """
          Combine two event checkers into an optional one.
  
          .. seealso:: :class:`OptionalTraceEventChecker`
          """
          return OptionalTraceEventChecker([self, other])
  
      @abc.abstractmethod
      def _str_internal(self, style=None, wrapped=True):
          """
          Format the boolean expression that this checker represents.
  
          :param style: When 'rst', a reStructuredText output is expected
          :type style: str
  
          :param wrapped: When True, the expression should be wrapped with
              parenthesis so it can be composed with other expressions.
          :type wrapped: bool
          """
  
          pass
  
      def doc_str(self):
          """
          Top-level function called by Sphinx's autodoc extension to augment
          docstrings of the functions.
          """
          return '\n    * {}'.format(self._str_internal(style='rst', wrapped=False))
  
      def __str__(self):
          return self._str_internal()
  
  
  class TraceEventChecker(TraceEventCheckerBase):
      """
      Check for one single event.
  
      :param event: Name of the event to check for.
      :type event: str
  
      :param check: Check that the listed events are present in the
          ``self.trace`` attribute of the instance on which the decorated
          method is applied.  If no such attribute is found, no check will be
          done.
      :type check: bool
      """
  
      def __init__(self, event, check=True):
          super().__init__(check=check)
          self.event = event
  
      def get_all_events(self):
          return {self.event}
  
      def check_events(self, event_set):
          if self.event not in event_set:
              raise MissingTraceEventError(self, available_events=event_set)
  
      def _str_internal(self, style=None, wrapped=True):
          template = '``{}``' if style == 'rst' else '{}'
          return template.format(self.event)
  
  
  class AssociativeTraceEventChecker(TraceEventCheckerBase):
      """
      Base class for associative operators like `and` and `or`
      """
  
      def __init__(self, op_str, event_checkers, check=True, prefix_str=''):
          super().__init__(check=check)
          checker_list = []
          optional_checker_list = []
          event_checkers = event_checkers or []
          for checker in event_checkers:
              # "unwrap" checkers of the same type, to avoid useless levels of
              # nesting. This is valid since the operator is known to be
              # associative. We don't use isinstance to avoid merging checkers
              # that may have different semantics.
              if type(checker) is type(self):
                  checker_list.extend(checker.checkers)
              # Aggregate them separately to avoid having multiple of them
              elif isinstance(checker, OptionalTraceEventChecker):
                  optional_checker_list.append(checker)
              else:
                  checker_list.append(checker)
  
          if optional_checker_list:
              checker_list.append(OptionalTraceEventChecker(optional_checker_list))
  
          # Avoid having the same event twice at the same level
          def key(checker):
              if isinstance(checker, TraceEventChecker):
                  return checker.event
              else:
                  return checker
          checker_list = deduplicate(checker_list, key=key)
  
          self.checkers = checker_list
          self.op_str = op_str
          self.prefix_str = prefix_str
  
      def get_all_events(self):
          events = set()
          for checker in self.checkers:
              events.update(checker.get_all_events())
          return events
  
      @classmethod
      def from_events(cls, events, **kwargs):
          """
          Build an instance of the class, converting ``str`` to
          ``TraceEventChecker``.
  
          :param events: Iterable of events
          :type events: list(str or TraceEventCheckerBase)
          """
          def make_event(e):
              if isinstance(e, TraceEventCheckerBase):
                  return e
              else:
                  return TraceEventChecker(e)
  
          return cls({
              make_event(e)
              for e in events
          }, **kwargs)
  
      def _str_internal(self, style=None, wrapped=True):
          op_str = ' {} '.format(self.op_str)
          unwrapped_str = self.prefix_str + op_str.join(
              c._str_internal(style=style, wrapped=True)
              # Sort for stable output
              for c in sorted(self.checkers, key=str)
          )
  
          template = '({})' if len(self.checkers) > 1 and wrapped else '{}'
          return template.format(unwrapped_str)
  
  
  class OrTraceEventChecker(AssociativeTraceEventChecker):
      """
      Check that one of the given event checkers is satisfied.
  
      :param event_checkers: Event checkers to check for
      :type event_checkers: list(TraceEventCheckerBase)
      """
  
      def __init__(self, event_checkers=None, **kwargs):
          super().__init__('or', event_checkers, **kwargs)
  
      def check_events(self, event_set):
          if not self.checkers:
              return
  
          failed_checker_set = set()
          for checker in self.checkers:
              try:
                  checker.check_events(event_set)
              except MissingTraceEventError as e:
                  failed_checker_set.add(e.missing_events)
              else:
                  break
          else:
              cls = type(self)
              raise MissingTraceEventError(
                  cls(failed_checker_set),
                  available_events=event_set,
              )
  
  
  class OptionalTraceEventChecker(AssociativeTraceEventChecker):
      """
      Do not check anything, but exposes the information that the events may be
      used if present.
  
      :param event_checkers: Event checkers that may be used
      :type event_checkers: list(TraceEventCheckerBase)
      """
  
      def __init__(self, event_checkers=None, **kwargs):
          super().__init__(',', event_checkers, prefix_str='optional: ', **kwargs)
  
      def check_events(self, event_set):
          return
  
  
  class AndTraceEventChecker(AssociativeTraceEventChecker):
      """
      Check that all the given event checkers are satisfied.
  
      :param event_checkers: Event checkers to check for
      :type event_checkers: list(TraceEventCheckerBase)
      """
  
      def __init__(self, event_checkers=None, **kwargs):
          super().__init__('and', event_checkers, **kwargs)
  
      def check_events(self, event_set):
          if not self.checkers:
              return
  
          failed_checker_set = set()
          for checker in self.checkers:
              try:
                  checker.check_events(event_set)
              except MissingTraceEventError as e:
                  failed_checker_set.add(e.missing_events)
  
          if failed_checker_set:
              cls = type(self)
              raise MissingTraceEventError(
                  cls(failed_checker_set),
                  available_events=event_set,
              )
  
      def doc_str(self):
          joiner = '\n' + '    '
          rst = joiner + joiner.join(
              '* {}'.format(c._str_internal(style='rst', wrapped=False))
              # Sort for stable output
              for c in sorted(self.checkers, key=str)
          )
          return rst
  
  
  def requires_events(*events, **kwargs):
      """
      Decorator for methods that require some given trace events.
  
      :param events: The list of required events
      :type events: list(str or TraceEventCheckerBase)
  
      :param check: Check that the listed events are present in the
          ``self.trace`` attribute of the instance on which the decorated method
          is applied.  If no such attribute is found, no check will be done.
      :type check: bool
  
      """
      return AndTraceEventChecker.from_events(events, **kwargs)
  
  
  def requires_one_event_of(*events, **kwargs):
      """
      Same as :func:`requires_events` with logical `OR` semantic.
      """
      return OrTraceEventChecker.from_events(events, **kwargs)
  
  
  def may_use_events(*events, **kwargs):
      """
      Same as :func:`requires_events` but just exposes some events that may be used
      if presents.
      """
      return OptionalTraceEventChecker.from_events(events, **kwargs)
  
  
  class MissingTraceEventError(RuntimeError, ValueError):
      """
      :param missing_events: The missing trace events
      :type missing_events: TraceEventCheckerBase
      """
  
      def __init__(self, missing_events, available_events=None):
          self.missing_events = missing_events
          # Forcibly turn into a list, to avoid carrying around an
          # _AvailableTraceEventsSet with its Trace instance
          self.available_events = sorted(available_events or [])
  
      def __str__(self):
          if self.available_events:
              available = '. Available events are: {}'.format(
                  ', '.join(sorted(self.available_events)))
          else:
              available = ''
  
          return "Trace is missing the following required events: {}{}".format(
              self.missing_events,
              available,
          )
          return msg
  
  
  class FtraceConf(SimpleMultiSrcConf, HideExekallID):
      """
      Configuration class of :class:`FtraceCollector`
  
      Available keys:
      {generated_help}
      """
      STRUCTURE = TopLevelKeyDesc('ftrace-conf', 'FTrace configuration', (
          KeyDesc('events', 'FTrace events to trace', [TypedList[str]]),
          KeyDesc('functions', 'FTrace functions to trace', [TypedList[str]]),
          KeyDesc('buffer-size', 'FTrace buffer size', [int]),
          KeyDesc('trace-clock', 'Clock used while tracing (see "trace_clock" in ftrace.txt kernel doc)', [str, None]),
          KeyDesc('saved-cmdlines-nr', 'Number of saved cmdlines with associated PID while tracing', [int]),
          KeyDesc('tracer', 'FTrace tracer to use', [str, None]),
      ))
  
      def add_merged_src(self, src, conf, **kwargs):
          """
          Merge-in a configuration source.
  
          :param src: Name of the merged source
          :type src: str
  
          :param conf: Conf to merge in
          :type conf: FtraceConf
          """
          def merge_conf(key, val):
  
              def non_mergeable(key):
                  if self.get(key, val) == val:
                      return val
                  else:
                      raise KeyError('Cannot merge key "{}": incompatible values specified: {} != {}'.format(
                          key, self[key], val,
                      ))
  
              if key in ('events', 'functions'):
                  return sorted(set(val) | set(self.get(key, [])))
              elif key == 'buffer-size':
                  return max(val, self.get(key, 0))
              elif key == 'trace-clock':
                  return non_mergeable(key)
              elif key == 'saved-cmdlines-nr':
                  return max(val, self.get(key, 0))
              elif key == 'tracer':
                  return non_mergeable(key)
              else:
                  raise KeyError('Cannot merge key "{}"'.format(key))
  
          merged = {
              key: merge_conf(key, val)
              for key, val in conf.items()
          }
  
          def is_modified(key, val):
              try:
                  existing_val = self[key]
              except KeyError:
                  return True
              else:
                  return val != existing_val
  
          # We merge some keys with their current value in the conf
          return self.add_src(src,
              conf={
                  key: val
                  for key, val in merged.items()
                  # Only add to the source if the result is different than what is
                  # already set
                  if is_modified(key, val)
              },
              **kwargs,
          )
  
  
  class CollectorBase(Loggable):
      """
      Base class for :class:`devlib.collector.CollectorBase`-based collectors
      using composition.
      """
  
      TOOLS = []
      """
      Sequence of tools to install on the target when using the collector.
      """
  
      def __init__(self, collector):
          self._collector = collector
          self._install_tools(collector.target)
  
      # Run once for each instance
      @lru_cache(maxsize=None)
      def _install_tools(self, target):
          target.install_tools(self.TOOLS)
  
      def __getattr__(self, attr):
          return getattr(self._collector, attr)
  
      def __enter__(self):
          return self._collector.__enter__()
  
      def __exit__(self, *args, **kwargs):
          return self._collector.__exit__(*args, **kwargs)
  
      def get_data(self, path):
          """
          Similar to :meth:`devlib.collector.CollectorBase.get_data` but takes
          the path directly as a parameter in order to disallow representing an
          invalid state where no path has been set.
          """
          coll = self._collector
          coll.set_output(path)
          return coll.get_data()
  
      @deprecate(replaced_by=get_data, deprecated_in='2.0', removed_in='2.1')
      def get_trace(self, path):
          """
          Deprecated alias for :meth:`get_data`.
          """
          return self.get_data(path)
  
  
  class FtraceCollector(CollectorBase, Configurable):
      """
      Thin wrapper around :class:`devlib.collector.ftrace.FtraceCollector`.
  
      {configurable_params}
      """
  
      CONF_CLASS = FtraceConf
      TOOLS = ['trace-cmd']
  
      def __init__(self, target, events=None, functions=None, buffer_size=10240, autoreport=False, trace_clock=None, saved_cmdlines_nr=8192, tracer=None, **kwargs):
          events = events or []
          functions = functions or []
          trace_clock = trace_clock or 'global'
          kwargs.update(
              target=target,
              events=events,
              functions=functions,
              buffer_size=buffer_size,
              autoreport=autoreport,
              trace_clock=trace_clock,
              saved_cmdlines_nr=saved_cmdlines_nr,
              tracer=tracer,
          )
          self.check_init_param(**kwargs)
  
          self.events = events
          kernel_events = set(itertools.chain.from_iterable(
              Trace.get_event_sources(event)
              for event in events
          ))
  
          # trace-cmd start complains if given these events, even though they are
          # valid
          kernel_events -= {
              'funcgraph_entry', 'funcgraph_exit',
              'print', 'bprint', 'bputs',
          }
  
          if 'funcgraph_entry' in events or 'funcgraph_exit' in events:
              tracer = 'function_graph' if tracer is None else tracer
  
          # Only pass true kernel events to devlib, as it will reject any other.
          kwargs.update(
              events=kernel_events,
              tracer=tracer,
          )
  
          # Install the tools before creating the collector, as devlib will check
          # for it
          self._install_tools(target)
          collector = devlib.FtraceCollector(
              # Prevent devlib from pushing its own trace-cmd since we provide
              # our own binary
              no_install=True,
              **kwargs
          )
          super().__init__(collector)
  
      @classmethod
      def from_conf(cls, target, conf):
          """
          Build an :class:`FtraceCollector` from a :class:`FtraceConf`
  
          :param target: Target to use when collecting the trace
          :type target: lisa.target.Target
  
          :param conf: Configuration object
          :type conf: FtraceConf
          """
          cls.get_logger().info('Ftrace configuration:\n{}'.format(conf))
          kwargs = cls.conf_to_init_kwargs(conf)
          kwargs['target'] = target
          cls.check_init_param(**kwargs)
          return cls(**kwargs)
  
      @classmethod
      def from_user_conf(cls, target, base_conf=None, user_conf=None, merged_src='merged'):
          """
          Build an :class:`FtraceCollector` from two :class:`FtraceConf`.
  
          ``base_conf`` is expected to contain the minimal configuration, and
          ``user_conf`` some additional settings that are used to augment the
          base configuration.
  
          :param target: Target to use when collecting the trace
          :type target: lisa.target.Target
  
          :param base_conf: Base configuration object, merged with ``user_conf``.
          :type base_conf: FtraceConf
  
          :param user_conf: User configuration object
          :type user_conf: FtraceConf
  
          :param merged_src: Name of the configuration source created by merging
              ``base_conf`` and ``user_conf``
          :type merged_src: str
          """
          user_conf = user_conf or FtraceConf()
          base_conf = base_conf or FtraceConf()
  
          # Make a copy of the conf, since it may be shared by multiple classes
          conf = copy.copy(base_conf)
  
          # Merge user configuration with the test's configuration
          conf.add_merged_src(
              src=merged_src,
              conf=user_conf,
          )
          return cls.from_conf(target, conf)
  
  
  class DmesgCollector(CollectorBase):
      """
      Wrapper around :class:`devlib.collector.dmesg.DmesgCollector`.
  
      It installs the ``dmesg`` tool automatically on the target upon creation,
      so we know what version is being is used.
      """
  
      TOOLS = ['dmesg']
      LOG_LEVELS = devlib.DmesgCollector.LOG_LEVELS
  
      def __init__(self, target, **kwargs):
          collector = devlib.DmesgCollector(target, **kwargs)
          super().__init__(collector)
  
  # vim :set tabstop=4 shiftwidth=4 expandtab textwidth=80
  