1 # The MIT License (MIT)
3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
5 # Permission is hereby granted, free of charge, to any person obtaining a copy
6 # of this software and associated documentation files (the "Software"), to deal
7 # in the Software without restriction, including without limitation the rights
8 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 # copies of the Software, and to permit persons to whom the Software is
10 # furnished to do so, subject to the following conditions:
12 # The above copyright notice and this permission notice shall be included in
13 # all copies or substantial portions of the Software.
15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
23 from bt2
import utils
, native_bt
26 from bt2
import message_iterator
as bt2_message_iterator
27 from bt2
import logging
as bt2_logging
28 from bt2
import port
as bt2_port
29 from bt2
import component
as bt2_component
30 from bt2
import value
as bt2_value
31 from bt2
import plugin
as bt2_plugin
33 from collections
import namedtuple
37 # a pair of component and ComponentSpec
38 _ComponentAndSpec
= namedtuple('_ComponentAndSpec', ['comp', 'spec'])
41 class _BaseComponentSpec
:
42 def __init__(self
, params
, obj
, logging_level
):
43 if logging_level
is not None:
44 utils
._check
_log
_level
(logging_level
)
46 self
._params
= bt2
.create_value(params
)
48 self
._logging
_level
= logging_level
59 def logging_level(self
):
60 return self
._logging
_level
63 class ComponentSpec(_BaseComponentSpec
):
70 logging_level
=bt2_logging
.LoggingLevel
.NONE
,
72 if type(params
) is str:
73 params
= {'inputs': [params
]}
75 super().__init
__(params
, obj
, logging_level
)
77 utils
._check
_str
(plugin_name
)
78 utils
._check
_str
(class_name
)
80 self
._plugin
_name
= plugin_name
81 self
._class
_name
= class_name
84 def plugin_name(self
):
85 return self
._plugin
_name
89 return self
._class
_name
92 class AutoSourceComponentSpec(_BaseComponentSpec
):
95 def __init__(self
, input, params
=None, obj
=_no_obj
, logging_level
=None):
96 super().__init
__(params
, obj
, logging_level
)
104 def _auto_discover_source_component_specs(auto_source_comp_specs
, plugin_set
):
105 # Transform a list of `AutoSourceComponentSpec` in a list of `ComponentSpec`
106 # using the automatic source discovery mechanism.
107 inputs
= bt2
.ArrayValue([spec
.input for spec
in auto_source_comp_specs
])
109 if plugin_set
is None:
110 plugin_set
= bt2
.find_plugins()
112 utils
._check
_type
(plugin_set
, bt2_plugin
._PluginSet
)
114 res_ptr
= native_bt
.bt2_auto_discover_source_components(
115 inputs
._ptr
, plugin_set
._ptr
119 raise bt2
._MemoryError('cannot auto discover source components')
121 res
= bt2_value
._create
_from
_ptr
(res_ptr
)
123 assert type(res
) == bt2
.MapValue
124 assert 'status' in res
126 status
= res
['status']
127 utils
._handle
_func
_status
(status
, 'cannot auto-discover source components')
130 comp_specs_raw
= res
['results']
131 assert type(comp_specs_raw
) == bt2
.ArrayValue
133 used_input_indices
= set()
135 for comp_spec_raw
in comp_specs_raw
:
136 assert type(comp_spec_raw
) == bt2
.ArrayValue
137 assert len(comp_spec_raw
) == 4
139 plugin_name
= comp_spec_raw
[0]
140 assert type(plugin_name
) == bt2
.StringValue
141 plugin_name
= str(plugin_name
)
143 class_name
= comp_spec_raw
[1]
144 assert type(class_name
) == bt2
.StringValue
145 class_name
= str(class_name
)
147 comp_inputs
= comp_spec_raw
[2]
148 assert type(comp_inputs
) == bt2
.ArrayValue
150 comp_orig_indices
= comp_spec_raw
[3]
151 assert type(comp_orig_indices
)
153 params
= bt2
.MapValue()
154 logging_level
= bt2
.LoggingLevel
.NONE
157 # Compute `params` for this component by piling up params given to all
158 # AutoSourceComponentSpec objects that contributed in the instantiation
161 # The effective log level for a component is the last one specified
162 # across the AutoSourceComponentSpec that contributed in its
164 for idx
in comp_orig_indices
:
165 orig_spec
= auto_source_comp_specs
[idx
]
167 if orig_spec
.params
is not None:
168 params
.update(orig_spec
.params
)
170 if orig_spec
.logging_level
is not None:
171 logging_level
= orig_spec
.logging_level
173 if orig_spec
.obj
is not AutoSourceComponentSpec
._no
_obj
:
176 used_input_indices
.add(int(idx
))
178 params
['inputs'] = comp_inputs
186 logging_level
=logging_level
,
190 if len(used_input_indices
) != len(inputs
):
191 unused_input_indices
= set(range(len(inputs
))) - used_input_indices
192 unused_input_indices
= sorted(unused_input_indices
)
193 unused_inputs
= [str(inputs
[x
]) for x
in unused_input_indices
]
196 'Some auto source component specs did not produce any component: '
197 + ', '.join(unused_inputs
)
199 raise RuntimeError(msg
)
204 # datetime.datetime or integral to nanoseconds
209 if isinstance(obj
, numbers
.Real
):
210 # consider that it's already in seconds
212 elif isinstance(obj
, datetime
.datetime
):
217 '"{}" is not an integral number or a datetime.datetime object'.format(obj
)
228 class _TraceCollectionMessageIteratorProxySink(bt2_component
._UserSinkComponent
):
229 def __init__(self
, params
, msg_list
):
230 assert type(msg_list
) is list
231 self
._msg
_list
= msg_list
232 self
._add
_input
_port
('in')
234 def _user_graph_is_configured(self
):
235 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
236 self
._input
_ports
['in']
239 def _user_consume(self
):
240 assert self
._msg
_list
[0] is None
241 self
._msg
_list
[0] = next(self
._msg
_iter
)
244 class TraceCollectionMessageIterator(bt2_message_iterator
._MessageIterator
):
247 source_component_specs
,
248 filter_component_specs
=None,
249 stream_intersection_mode
=False,
254 utils
._check
_bool
(stream_intersection_mode
)
255 self
._stream
_intersection
_mode
= stream_intersection_mode
256 self
._begin
_ns
= _get_ns(begin
)
257 self
._end
_ns
= _get_ns(end
)
258 self
._msg
_list
= [None]
260 # If a single item is provided, convert to a list.
261 if type(source_component_specs
) in (
263 AutoSourceComponentSpec
,
266 source_component_specs
= [source_component_specs
]
268 # Convert any string to an AutoSourceComponentSpec.
269 def str_to_auto(item
):
270 if type(item
) is str:
271 item
= AutoSourceComponentSpec(item
)
275 source_component_specs
= [str_to_auto(s
) for s
in source_component_specs
]
277 if type(filter_component_specs
) is ComponentSpec
:
278 filter_component_specs
= [filter_component_specs
]
279 elif filter_component_specs
is None:
280 filter_component_specs
= []
282 self
._validate
_source
_component
_specs
(source_component_specs
)
283 self
._validate
_filter
_component
_specs
(filter_component_specs
)
285 # Pass any `ComponentSpec` instance as-is.
286 self
._src
_comp
_specs
= [
287 spec
for spec
in source_component_specs
if type(spec
) is ComponentSpec
290 # Convert any `AutoSourceComponentSpec` in concrete `ComponentSpec` instances.
291 auto_src_comp_specs
= [
293 for spec
in source_component_specs
294 if type(spec
) is AutoSourceComponentSpec
296 self
._src
_comp
_specs
+= _auto_discover_source_component_specs(
297 auto_src_comp_specs
, plugin_set
300 self
._flt
_comp
_specs
= filter_component_specs
301 self
._next
_suffix
= 1
302 self
._connect
_ports
= False
304 # lists of _ComponentAndSpec
305 self
._src
_comps
_and
_specs
= []
306 self
._flt
_comps
_and
_specs
= []
310 def _validate_source_component_specs(self
, comp_specs
):
311 for comp_spec
in comp_specs
:
313 type(comp_spec
) is not ComponentSpec
314 and type(comp_spec
) is not AutoSourceComponentSpec
317 '"{}" object is not a ComponentSpec or AutoSourceComponentSpec'.format(
322 def _validate_filter_component_specs(self
, comp_specs
):
323 for comp_spec
in comp_specs
:
324 if type(comp_spec
) is not ComponentSpec
:
326 '"{}" object is not a ComponentSpec'.format(type(comp_spec
))
330 assert self
._msg
_list
[0] is None
331 self
._graph
.run_once()
332 msg
= self
._msg
_list
[0]
333 assert msg
is not None
334 self
._msg
_list
[0] = None
337 def _create_stream_intersection_trimmer(self
, component
, port
):
338 # find the original parameters specified by the user to create
339 # this port's component to get the `inputs` parameter
340 for src_comp_and_spec
in self
._src
_comps
_and
_specs
:
341 if component
== src_comp_and_spec
.comp
:
345 inputs
= src_comp_and_spec
.spec
.params
['inputs']
346 except Exception as e
:
348 'all source components must be created with an "inputs" parameter in stream intersection mode'
351 params
= {'inputs': inputs
}
353 # query the port's component for the `babeltrace.trace-info`
354 # object which contains the stream intersection range for each
356 query_exec
= bt2
.QueryExecutor(
357 src_comp_and_spec
.comp
.cls
, 'babeltrace.trace-info', params
359 trace_info_res
= query_exec
.query()
363 # find the trace info for this port's trace
365 for trace_info
in trace_info_res
:
366 for stream
in trace_info
['streams']:
367 if stream
['port-name'] == port
.name
:
368 range_ns
= trace_info
['intersection-range-ns']
369 begin
= range_ns
['begin']
370 end
= range_ns
['end']
375 if begin
is None or end
is None:
377 'cannot find stream intersection range for port "{}"'.format(port
.name
)
380 name
= 'trimmer-{}-{}'.format(src_comp_and_spec
.comp
.name
, port
.name
)
381 return self
._create
_trimmer
(begin
, end
, name
)
383 def _create_muxer(self
):
384 plugin
= bt2
.find_plugin('utils')
387 raise RuntimeError('cannot find "utils" plugin (needed for the muxer)')
389 if 'muxer' not in plugin
.filter_component_classes
:
391 'cannot find "muxer" filter component class in "utils" plugin'
394 comp_cls
= plugin
.filter_component_classes
['muxer']
395 return self
._graph
.add_component(comp_cls
, 'muxer')
397 def _create_trimmer(self
, begin_ns
, end_ns
, name
):
398 plugin
= bt2
.find_plugin('utils')
401 raise RuntimeError('cannot find "utils" plugin (needed for the trimmer)')
403 if 'trimmer' not in plugin
.filter_component_classes
:
405 'cannot find "trimmer" filter component class in "utils" plugin'
410 def ns_to_string(ns
):
411 s_part
= ns
// 1000000000
412 ns_part
= ns
% 1000000000
413 return '{}.{:09d}'.format(s_part
, ns_part
)
415 if begin_ns
is not None:
416 params
['begin'] = ns_to_string(begin_ns
)
418 if end_ns
is not None:
419 params
['end'] = ns_to_string(end_ns
)
421 comp_cls
= plugin
.filter_component_classes
['trimmer']
422 return self
._graph
.add_component(comp_cls
, name
, params
)
424 def _get_unique_comp_name(self
, comp_spec
):
425 name
= '{}-{}'.format(comp_spec
.plugin_name
, comp_spec
.class_name
)
426 comps_and_specs
= itertools
.chain(
427 self
._src
_comps
_and
_specs
, self
._flt
_comps
_and
_specs
430 if name
in [comp_and_spec
.comp
.name
for comp_and_spec
in comps_and_specs
]:
431 name
+= '-{}'.format(self
._next
_suffix
)
432 self
._next
_suffix
+= 1
436 def _component_spec_class(self
, comp_spec
, comp_cls_type
):
437 plugin
= bt2
.find_plugin(comp_spec
.plugin_name
)
440 raise ValueError('no such plugin: {}'.format(comp_spec
.plugin_name
))
442 if comp_cls_type
== _CompClsType
.SOURCE
:
443 comp_classes
= plugin
.source_component_classes
445 comp_classes
= plugin
.filter_component_classes
447 if comp_spec
.class_name
not in comp_classes
:
448 cc_type
= 'source' if comp_cls_type
== _CompClsType
.SOURCE
else 'filter'
450 'no such {} component class in "{}" plugin: {}'.format(
451 cc_type
, comp_spec
.plugin_name
, comp_spec
.class_name
455 return comp_classes
[comp_spec
.class_name
]
457 def _create_comp(self
, comp_spec
, comp_cls_type
):
458 comp_cls
= self
._component
_spec
_class
(comp_spec
, comp_cls_type
)
459 name
= self
._get
_unique
_comp
_name
(comp_spec
)
460 comp
= self
._graph
.add_component(
461 comp_cls
, name
, comp_spec
.params
, comp_spec
.obj
, comp_spec
.logging_level
465 def _get_free_muxer_input_port(self
):
466 for port
in self
._muxer
_comp
.input_ports
.values():
467 if not port
.is_connected
:
470 def _connect_src_comp_port(self
, component
, port
):
471 # if this trace collection iterator is in stream intersection
472 # mode, we need this connection:
474 # port -> trimmer -> muxer
479 if self
._stream
_intersection
_mode
:
480 trimmer_comp
= self
._create
_stream
_intersection
_trimmer
(component
, port
)
481 self
._graph
.connect_ports(port
, trimmer_comp
.input_ports
['in'])
482 port_to_muxer
= trimmer_comp
.output_ports
['out']
486 self
._graph
.connect_ports(port_to_muxer
, self
._get
_free
_muxer
_input
_port
())
488 def _graph_port_added(self
, component
, port
):
489 if not self
._connect
_ports
:
492 if type(port
) is bt2_port
._InputPort
:
495 if component
not in [comp
.comp
for comp
in self
._src
_comps
_and
_specs
]:
496 # do not care about non-source components (muxer, trimmer, etc.)
499 self
._connect
_src
_comp
_port
(component
, port
)
501 def _get_greatest_operative_mip_version(self
):
502 def append_comp_specs_descriptors(descriptors
, comp_specs
, comp_cls_type
):
503 for comp_spec
in comp_specs
:
504 comp_cls
= self
._component
_spec
_class
(comp_spec
, comp_cls_type
)
506 bt2
.ComponentDescriptor(comp_cls
, comp_spec
.params
, comp_spec
.obj
)
510 append_comp_specs_descriptors(
511 descriptors
, self
._src
_comp
_specs
, _CompClsType
.SOURCE
513 append_comp_specs_descriptors(
514 descriptors
, self
._flt
_comp
_specs
, _CompClsType
.FILTER
517 if self
._stream
_intersection
_mode
:
518 # we also need at least one `flt.utils.trimmer` component
519 comp_spec
= ComponentSpec('utils', 'trimmer')
520 append_comp_specs_descriptors(descriptors
, [comp_spec
], _CompClsType
.FILTER
)
522 mip_version
= bt2
.get_greatest_operative_mip_version(descriptors
)
524 if mip_version
is None:
525 msg
= 'failed to find an operative message interchange protocol version (components are not interoperable)'
526 raise RuntimeError(msg
)
530 def _build_graph(self
):
531 self
._graph
= bt2
.Graph(self
._get
_greatest
_operative
_mip
_version
())
532 self
._graph
.add_port_added_listener(self
._graph
_port
_added
)
533 self
._muxer
_comp
= self
._create
_muxer
()
535 if self
._begin
_ns
is not None or self
._end
_ns
is not None:
536 trimmer_comp
= self
._create
_trimmer
(self
._begin
_ns
, self
._end
_ns
, 'trimmer')
537 self
._graph
.connect_ports(
538 self
._muxer
_comp
.output_ports
['out'], trimmer_comp
.input_ports
['in']
540 last_flt_out_port
= trimmer_comp
.output_ports
['out']
542 last_flt_out_port
= self
._muxer
_comp
.output_ports
['out']
544 # create extra filter components (chained)
545 for comp_spec
in self
._flt
_comp
_specs
:
546 comp
= self
._create
_comp
(comp_spec
, _CompClsType
.FILTER
)
547 self
._flt
_comps
_and
_specs
.append(_ComponentAndSpec(comp
, comp_spec
))
549 # connect the extra filter chain
550 for comp_and_spec
in self
._flt
_comps
_and
_specs
:
551 in_port
= list(comp_and_spec
.comp
.input_ports
.values())[0]
552 out_port
= list(comp_and_spec
.comp
.output_ports
.values())[0]
553 self
._graph
.connect_ports(last_flt_out_port
, in_port
)
554 last_flt_out_port
= out_port
556 # Here we create the components, self._graph_port_added() is
557 # called when they add ports, but the callback returns early
558 # because self._connect_ports is False. This is because the
559 # self._graph_port_added() could not find the associated source
560 # component specification in self._src_comps_and_specs because
561 # it does not exist yet (it needs the created component to
563 for comp_spec
in self
._src
_comp
_specs
:
564 comp
= self
._create
_comp
(comp_spec
, _CompClsType
.SOURCE
)
565 self
._src
_comps
_and
_specs
.append(_ComponentAndSpec(comp
, comp_spec
))
567 # Now we connect the ports which exist at this point. We allow
568 # self._graph_port_added() to automatically connect _new_ ports.
569 self
._connect
_ports
= True
571 for comp_and_spec
in self
._src
_comps
_and
_specs
:
572 # Keep a separate list because comp_and_spec.output_ports
573 # could change during the connection of one of its ports.
574 # Any new port is handled by self._graph_port_added().
575 out_ports
= [port
for port
in comp_and_spec
.comp
.output_ports
.values()]
577 for out_port
in out_ports
:
578 if out_port
.is_connected
:
581 self
._connect
_src
_comp
_port
(comp_and_spec
.comp
, out_port
)
583 # Add the proxy sink, passing our message list to share consumed
584 # messages with this trace collection message iterator.
585 sink
= self
._graph
.add_component(
586 _TraceCollectionMessageIteratorProxySink
, 'proxy-sink', obj
=self
._msg
_list
588 sink_in_port
= sink
.input_ports
['in']
590 # connect last filter to proxy sink
591 self
._graph
.connect_ports(last_flt_out_port
, sink_in_port
)