From 85dcce24a87eb7269f0f4e2d21d4e049f4c9a0ee Mon Sep 17 00:00:00 2001 From: Philippe Proulx Date: Fri, 15 Sep 2017 19:18:24 -0400 Subject: [PATCH] bt2: add trace collection notification iterator MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit A trace collection notification iterator is a special notification iterator which iterates on the multiplexed notifications of all the output ports of one or more traces. The new notification iterator supports a stream intersection mode and trimming the whole multiplexed stream. Signed-off-by: Philippe Proulx Signed-off-by: Jérémie Galarneau --- bindings/python/bt2/Makefile.am | 109 +++---- bindings/python/bt2/bt2/__init__.py.in | 1 + .../trace_collection_notification_iterator.py | 280 ++++++++++++++++++ 3 files changed, 336 insertions(+), 54 deletions(-) create mode 100644 bindings/python/bt2/bt2/trace_collection_notification_iterator.py diff --git a/bindings/python/bt2/Makefile.am b/bindings/python/bt2/Makefile.am index f8ae782a..fe1c83b1 100644 --- a/bindings/python/bt2/Makefile.am +++ b/bindings/python/bt2/Makefile.am @@ -5,62 +5,63 @@ AM_LDFLAGS=-L$(top_builddir)/lib/.libs INSTALLED_FILES=$(builddir)/installed_files.txt -STATIC_BINDINGS_DEPS = \ - bt2/logging.c \ - bt2/logging.h \ - bt2/native_btccpriomap.i \ - bt2/native_btclockclass.i \ - bt2/native_btcomponentclass.i \ - bt2/native_btcomponent.i \ - bt2/native_btconnection.i \ - bt2/native_btctfwriter.i \ - bt2/native_bteventclass.i \ - bt2/native_btevent.i \ - bt2/native_btfields.i \ - bt2/native_btft.i \ - bt2/native_btgraph.i \ - bt2/native_bt.i \ - bt2/native_btlogging.i \ - bt2/native_btnotification.i \ - bt2/native_btnotifiter.i \ - bt2/native_btpacket.i \ - bt2/native_btplugin.i \ - bt2/native_btport.i \ - bt2/native_btqueryexec.i \ - bt2/native_btref.i \ - bt2/native_btstreamclass.i \ - bt2/native_btstream.i \ - bt2/native_bttrace.i \ - bt2/native_btvalues.i \ - bt2/native_btversion.i \ - bt2/clock_class_priority_map.py \ - bt2/clock_class.py \ - bt2/clock_value.py \ - bt2/component.py \ - bt2/connection.py \ - bt2/ctf_writer.py \ - bt2/event_class.py \ - bt2/event.py \ - bt2/fields.py \ - bt2/field_types.py \ - bt2/graph.py \ - bt2/logging.py \ - bt2/notification_iterator.py \ - bt2/notification.py \ - bt2/object.py \ - bt2/packet.py \ - bt2/plugin.py \ - bt2/port.py \ - bt2/py_plugin.py \ - bt2/query_executor.py \ - bt2/stream_class.py \ - bt2/stream.py \ - bt2/trace.py \ - bt2/utils.py \ +STATIC_BINDINGS_DEPS = \ + bt2/logging.c \ + bt2/logging.h \ + bt2/native_btccpriomap.i \ + bt2/native_btclockclass.i \ + bt2/native_btcomponentclass.i \ + bt2/native_btcomponent.i \ + bt2/native_btconnection.i \ + bt2/native_btctfwriter.i \ + bt2/native_bteventclass.i \ + bt2/native_btevent.i \ + bt2/native_btfields.i \ + bt2/native_btft.i \ + bt2/native_btgraph.i \ + bt2/native_bt.i \ + bt2/native_btlogging.i \ + bt2/native_btnotification.i \ + bt2/native_btnotifiter.i \ + bt2/native_btpacket.i \ + bt2/native_btplugin.i \ + bt2/native_btport.i \ + bt2/native_btqueryexec.i \ + bt2/native_btref.i \ + bt2/native_btstreamclass.i \ + bt2/native_btstream.i \ + bt2/native_bttrace.i \ + bt2/native_btvalues.i \ + bt2/native_btversion.i \ + bt2/clock_class_priority_map.py \ + bt2/clock_class.py \ + bt2/clock_value.py \ + bt2/component.py \ + bt2/connection.py \ + bt2/ctf_writer.py \ + bt2/event_class.py \ + bt2/event.py \ + bt2/fields.py \ + bt2/field_types.py \ + bt2/graph.py \ + bt2/logging.py \ + bt2/notification_iterator.py \ + bt2/notification.py \ + bt2/object.py \ + bt2/packet.py \ + bt2/plugin.py \ + bt2/port.py \ + bt2/py_plugin.py \ + bt2/query_executor.py \ + bt2/stream_class.py \ + bt2/stream.py \ + bt2/trace.py \ + bt2/trace_collection_notification_iterator.py \ + bt2/utils.py \ bt2/values.py -GENERATED_BINDINGS_DEPS = \ - bt2/__init__.py \ +GENERATED_BINDINGS_DEPS = \ + bt2/__init__.py \ setup.py BUILD_FLAGS=CC="$(CC)" \ diff --git a/bindings/python/bt2/bt2/__init__.py.in b/bindings/python/bt2/bt2/__init__.py.in index b86fe750..554391ec 100644 --- a/bindings/python/bt2/bt2/__init__.py.in +++ b/bindings/python/bt2/bt2/__init__.py.in @@ -76,6 +76,7 @@ from bt2.query_executor import * from bt2.stream import _Stream from bt2.stream_class import * from bt2.trace import * +from bt2.trace_collection_notification_iterator import * from bt2.values import * from bt2.values import _Value diff --git a/bindings/python/bt2/bt2/trace_collection_notification_iterator.py b/bindings/python/bt2/bt2/trace_collection_notification_iterator.py new file mode 100644 index 00000000..179ac357 --- /dev/null +++ b/bindings/python/bt2/bt2/trace_collection_notification_iterator.py @@ -0,0 +1,280 @@ +# The MIT License (MIT) +# +# Copyright (c) 2017 Philippe Proulx +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. + +from bt2 import utils +import bt2 +import bt2.notification_iterator +import datetime +import collections.abc +from collections import namedtuple +import numbers + + +# a pair of source component and _SourceComponentSpec +_SourceComponentAndSpec = namedtuple('_SourceComponentAndSpec', + ['comp', 'spec']) + + +class SourceComponentSpec: + def __init__(self, plugin_name, component_class_name, params=None): + utils._check_str(plugin_name) + utils._check_str(component_class_name) + self._plugin_name = plugin_name + self._component_class_name = component_class_name + + if type(params) is str: + self._params = bt2.create_value({'path': params}) + else: + self._params = bt2.create_value(params) + + @property + def plugin_name(self): + return self._plugin_name + + @property + def component_class_name(self): + return self._component_class_name + + @property + def params(self): + return self._params + + +# datetime.datetime or integral to nanoseconds +def _get_ns(obj): + if obj is None: + return + + if isinstance(obj, numbers.Real): + # consider that it's already in seconds + s = obj + elif isinstance(obj, datetime.datetime): + # s -> ns + s = obj.timestamp() + else: + raise TypeError('"{}" is not an integral number or a datetime.datetime object'.format(obj)) + + return int(s * 1e9) + + +class TraceCollectionNotificationIterator(bt2.notification_iterator._NotificationIterator): + def __init__(self, source_component_specs, notification_types=None, + stream_intersection_mode=False, begin=None, + end=None): + utils._check_bool(stream_intersection_mode) + self._stream_intersection_mode = stream_intersection_mode + self._begin_ns = _get_ns(begin) + self._end_ns = _get_ns(end) + self._notification_types = notification_types + self._src_comp_specs = source_component_specs + self._next_suffix = 1 + self._connect_ports = False + + # set of _SourceComponentAndSpec + self._src_comps_and_specs = [] + + self._validate_source_component_specs() + self._build_graph() + + def _validate_source_component_specs(self): + for source_comp_spec in self._src_comp_specs: + if type(source_comp_spec) is not SourceComponentSpec: + raise TypeError('"{}" object is not a SourceComponentSpec'.format(type(source_comp_spec))) + + def __next__(self): + return next(self._notif_iter) + + def _create_stream_intersection_trimmer(self, port): + # find the original parameters specified by the user to create + # this port's component to get the `path` parameter + for src_comp_and_spec in self._src_comps_and_specs: + if port.component == src_comp_and_spec.comp: + params = src_comp_and_spec.spec.params + break + + try: + path = params['path'] + except: + raise bt2.Error('all source components must be created with a "path" parameter in stream intersection mode') + + params = {'path': str(path)} + + # query the port's component for the `trace-info` object which + # contains the stream intersection range for each exposed + # trace + query_exec = bt2.QueryExecutor() + trace_info_res = query_exec.query(port.component.component_class, + 'trace-info', params) + begin = None + end = None + + # find the trace info for this port's trace by name's prefix + for trace_info in trace_info_res: + if port.name.startswith(str(trace_info['path'])): + range_ns = trace_info['intersection-range-ns'] + begin = range_ns['begin'] + end = range_ns['end'] + break + + if begin is None or end is None: + raise bt2.Error('cannot find stream intersection range for port "{}"'.format(port.name)) + + name = 'trimmer-{}-{}'.format(port.component.name, port.name) + return self._create_trimmer(begin, end, name) + + def _create_muxer(self): + plugin = bt2.find_plugin('utils') + + if plugin is None: + raise bt2.Error('cannot find "utils" plugin (needed for the muxer)') + + if 'muxer' not in plugin.filter_component_classes: + raise bt2.Error('cannot find "muxer" filter component class in "utils" plugin') + + comp_cls = plugin.filter_component_classes['muxer'] + return self._graph.add_component(comp_cls, 'muxer') + + def _create_trimmer(self, begin, end, name): + plugin = bt2.find_plugin('utils') + + if plugin is None: + raise bt2.Error('cannot find "utils" plugin (needed for the trimmer)') + + if 'trimmer' not in plugin.filter_component_classes: + raise bt2.Error('cannot find "trimmer" filter component class in "utils" plugin') + + params = {} + + if begin is not None: + params['begin'] = begin + + if end is not None: + params['end'] = end + + comp_cls = plugin.filter_component_classes['trimmer'] + return self._graph.add_component(comp_cls, name, params) + + def _get_unique_src_comp_name(self, comp_spec): + name = '{}-{}'.format(comp_spec.plugin_name, + comp_spec.component_class_name) + + if name in [comp_and_spec.comp.name for comp_and_spec in self._src_comps_and_specs]: + name += '-{}'.format(self._next_suffix) + self._next_suffix += 1 + + return name + + def _create_src_comp(self, comp_spec): + plugin = bt2.find_plugin(comp_spec.plugin_name) + + if plugin is None: + raise bt2.Error('no such plugin: {}'.format(comp_spec.plugin_name)) + + if comp_spec.component_class_name not in plugin.source_component_classes: + raise bt2.Error('no such source component class in "{}" plugin: {}'.format(comp_spec.plugin_name, + comp_spec.component_class_name)) + + comp_cls = plugin.source_component_classes[comp_spec.component_class_name] + name = self._get_unique_src_comp_name(comp_spec) + comp = self._graph.add_component(comp_cls, name, comp_spec.params) + return comp + + def _get_free_muxer_input_port(self): + for port in self._muxer_comp.input_ports.values(): + if not port.is_connected: + return port + + def _connect_src_comp_port(self, port): + # if this trace collection iterator is in stream intersection + # mode, we need this connection: + # + # port -> trimmer -> muxer + # + # otherwise, simply: + # + # port -> muxer + if self._stream_intersection_mode: + trimmer_comp = self._create_stream_intersection_trimmer(port) + self._graph.connect_ports(port, trimmer_comp.input_ports['in']) + port_to_muxer = trimmer_comp.output_ports['out'] + else: + port_to_muxer = port + + self._graph.connect_ports(port_to_muxer, self._get_free_muxer_input_port()) + + def _graph_port_added(self, port): + if not self._connect_ports: + return + + if type(port) is bt2._InputPort: + return + + if port.component not in [comp.comp for comp in self._src_comps_and_specs]: + # do not care about non-source components (muxer, trimmer, etc.) + return + + self._connect_src_comp_port(port) + + def _build_graph(self): + self._graph = bt2.Graph() + self._graph.add_listener(bt2.GraphListenerType.PORT_ADDED, + self._graph_port_added) + self._muxer_comp = self._create_muxer() + + if self._begin_ns is not None or self._end_ns is not None: + trimmer_comp = self._create_trimmer(self._begin_ns, + self._end_ns, 'trimmer') + self._graph.connect_ports(self._muxer_comp.output_ports['out'], + trimmer_comp.input_ports['in']) + notif_iter_port = trimmer_comp.output_ports['out'] + else: + notif_iter_port = self._muxer_comp.output_ports['out'] + + # Here we create the components, self._graph_port_added() is + # called when they add ports, but the callback returns early + # because self._connect_ports is False. This is because the + # self._graph_port_added() could not find the associated source + # component specification in self._src_comps_and_specs because + # it does not exist yet (it needs the created component to + # exist). + for comp_spec in self._src_comp_specs: + comp = self._create_src_comp(comp_spec) + self._src_comps_and_specs.append(_SourceComponentAndSpec(comp, comp_spec)) + + # Now we connect the ports which exist at this point. We allow + # self._graph_port_added() to automatically connect _new_ ports. + self._connect_ports = True + + for comp_and_spec in self._src_comps_and_specs: + # Keep a separate list because comp_and_spec.output_ports + # could change during the connection of one of its ports. + # Any new port is handled by self._graph_port_added(). + out_ports = [port for port in comp_and_spec.comp.output_ports.values()] + + for out_port in out_ports: + if not out_port.component or out_port.is_connected: + continue + + self._connect_src_comp_port(out_port) + + # create this trace collection iterator's notification iterator + self._notif_iter = notif_iter_port.create_notification_iterator(self._notification_types) -- 2.34.1