1 # SPDX-License-Identifier: MIT
3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
5 from bt2
import utils
, native_bt
8 from bt2
import message_iterator
as bt2_message_iterator
9 from bt2
import port
as bt2_port
10 from bt2
import component
as bt2_component
11 from bt2
import value
as bt2_value
12 from bt2
import plugin
as bt2_plugin
14 from collections
import namedtuple
18 # a pair of component and ComponentSpec
19 _ComponentAndSpec
= namedtuple('_ComponentAndSpec', ['comp', 'spec'])
22 class _BaseComponentSpec
:
23 # Base for any component spec that can be passed to
24 # TraceCollectionMessageIterator.
25 def __init__(self
, params
, obj
, logging_level
):
26 if logging_level
is not None:
27 utils
._check
_log
_level
(logging_level
)
29 self
._params
= bt2
.create_value(params
)
31 self
._logging
_level
= logging_level
42 def logging_level(self
):
43 return self
._logging
_level
46 class ComponentSpec(_BaseComponentSpec
):
47 # A component spec with a specific component class.
53 logging_level
=bt2
.LoggingLevel
.NONE
,
55 if type(params
) is str:
56 params
= {'inputs': [params
]}
58 super().__init
__(params
, obj
, logging_level
)
60 is_cc_object
= isinstance(
62 (bt2
._SourceComponentClassConst
, bt2
._FilterComponentClassConst
),
64 is_user_cc_type
= isinstance(
65 component_class
, bt2_component
._UserComponentType
67 component_class
, (bt2
._UserSourceComponent
, bt2
._UserFilterComponent
)
70 if not is_cc_object
and not is_user_cc_type
:
72 "'{}' is not a source or filter component class".format(
73 component_class
.__class
__.__name
__
77 self
._component
_class
= component_class
80 def component_class(self
):
81 return self
._component
_class
84 def from_named_plugin_and_component_class(
90 logging_level
=bt2
.LoggingLevel
.NONE
,
92 plugin
= bt2
.find_plugin(plugin_name
)
95 raise ValueError('no such plugin: {}'.format(plugin_name
))
97 if component_class_name
in plugin
.source_component_classes
:
98 comp_class
= plugin
.source_component_classes
[component_class_name
]
99 elif component_class_name
in plugin
.filter_component_classes
:
100 comp_class
= plugin
.filter_component_classes
[component_class_name
]
103 'source or filter component class `{}` not found in plugin `{}`'.format(
104 component_class_name
, plugin_name
108 return cls(comp_class
, params
, obj
, logging_level
)
111 class AutoSourceComponentSpec(_BaseComponentSpec
):
112 # A component spec that does automatic source discovery.
115 def __init__(self
, input, params
=None, obj
=_no_obj
, logging_level
=None):
116 super().__init
__(params
, obj
, logging_level
)
124 def _auto_discover_source_component_specs(auto_source_comp_specs
, plugin_set
):
125 # Transform a list of `AutoSourceComponentSpec` in a list of `ComponentSpec`
126 # using the automatic source discovery mechanism.
127 inputs
= bt2
.ArrayValue([spec
.input for spec
in auto_source_comp_specs
])
129 if plugin_set
is None:
130 plugin_set
= bt2
.find_plugins()
132 utils
._check
_type
(plugin_set
, bt2_plugin
._PluginSet
)
134 res_ptr
= native_bt
.bt2_auto_discover_source_components(
135 inputs
._ptr
, plugin_set
._ptr
139 raise bt2
._MemoryError('cannot auto discover source components')
141 res
= bt2_value
._create
_from
_ptr
(res_ptr
)
143 assert type(res
) == bt2
.MapValue
144 assert 'status' in res
146 status
= res
['status']
147 utils
._handle
_func
_status
(status
, 'cannot auto-discover source components')
150 comp_specs_raw
= res
['results']
151 assert type(comp_specs_raw
) == bt2
.ArrayValue
153 used_input_indices
= set()
155 for comp_spec_raw
in comp_specs_raw
:
156 assert type(comp_spec_raw
) == bt2
.ArrayValue
157 assert len(comp_spec_raw
) == 4
159 plugin_name
= comp_spec_raw
[0]
160 assert type(plugin_name
) == bt2
.StringValue
161 plugin_name
= str(plugin_name
)
163 class_name
= comp_spec_raw
[1]
164 assert type(class_name
) == bt2
.StringValue
165 class_name
= str(class_name
)
167 comp_inputs
= comp_spec_raw
[2]
168 assert type(comp_inputs
) == bt2
.ArrayValue
170 comp_orig_indices
= comp_spec_raw
[3]
171 assert type(comp_orig_indices
)
173 params
= bt2
.MapValue()
174 logging_level
= bt2
.LoggingLevel
.NONE
177 # Compute `params` for this component by piling up params given to all
178 # AutoSourceComponentSpec objects that contributed in the instantiation
181 # The effective log level for a component is the last one specified
182 # across the AutoSourceComponentSpec that contributed in its
184 for idx
in comp_orig_indices
:
185 orig_spec
= auto_source_comp_specs
[idx
]
187 if orig_spec
.params
is not None:
188 params
.update(orig_spec
.params
)
190 if orig_spec
.logging_level
is not None:
191 logging_level
= orig_spec
.logging_level
193 if orig_spec
.obj
is not AutoSourceComponentSpec
._no
_obj
:
196 used_input_indices
.add(int(idx
))
198 params
['inputs'] = comp_inputs
201 ComponentSpec
.from_named_plugin_and_component_class(
206 logging_level
=logging_level
,
210 if len(used_input_indices
) != len(inputs
):
211 unused_input_indices
= set(range(len(inputs
))) - used_input_indices
212 unused_input_indices
= sorted(unused_input_indices
)
213 unused_inputs
= [str(inputs
[x
]) for x
in unused_input_indices
]
216 'Some auto source component specs did not produce any component: '
217 + ', '.join(unused_inputs
)
219 raise RuntimeError(msg
)
224 # datetime.datetime or integral to nanoseconds
229 if isinstance(obj
, numbers
.Real
):
230 # consider that it's already in seconds
232 elif isinstance(obj
, datetime
.datetime
):
237 '"{}" is not an integral number or a datetime.datetime object'.format(obj
)
243 class _TraceCollectionMessageIteratorProxySink(bt2_component
._UserSinkComponent
):
244 def __init__(self
, config
, params
, msg_list
):
245 assert type(msg_list
) is list
246 self
._msg
_list
= msg_list
247 self
._add
_input
_port
('in')
249 def _user_graph_is_configured(self
):
250 self
._msg
_iter
= self
._create
_message
_iterator
(self
._input
_ports
['in'])
252 def _user_consume(self
):
253 assert self
._msg
_list
[0] is None
254 self
._msg
_list
[0] = next(self
._msg
_iter
)
257 class TraceCollectionMessageIterator(bt2_message_iterator
._MessageIterator
):
260 source_component_specs
,
261 filter_component_specs
=None,
262 stream_intersection_mode
=False,
267 utils
._check
_bool
(stream_intersection_mode
)
268 self
._stream
_intersection
_mode
= stream_intersection_mode
269 self
._begin
_ns
= _get_ns(begin
)
270 self
._end
_ns
= _get_ns(end
)
271 self
._msg
_list
= [None]
273 # If a single item is provided, convert to a list.
274 if type(source_component_specs
) in (
276 AutoSourceComponentSpec
,
279 source_component_specs
= [source_component_specs
]
281 # Convert any string to an AutoSourceComponentSpec.
282 def str_to_auto(item
):
283 if type(item
) is str:
284 item
= AutoSourceComponentSpec(item
)
288 source_component_specs
= [str_to_auto(s
) for s
in source_component_specs
]
290 if type(filter_component_specs
) is ComponentSpec
:
291 filter_component_specs
= [filter_component_specs
]
292 elif filter_component_specs
is None:
293 filter_component_specs
= []
295 self
._validate
_source
_component
_specs
(source_component_specs
)
296 self
._validate
_filter
_component
_specs
(filter_component_specs
)
298 # Pass any `ComponentSpec` instance as-is.
299 self
._src
_comp
_specs
= [
300 spec
for spec
in source_component_specs
if type(spec
) is ComponentSpec
303 # Convert any `AutoSourceComponentSpec` in concrete `ComponentSpec` instances.
304 auto_src_comp_specs
= [
306 for spec
in source_component_specs
307 if type(spec
) is AutoSourceComponentSpec
309 self
._src
_comp
_specs
+= _auto_discover_source_component_specs(
310 auto_src_comp_specs
, plugin_set
313 self
._flt
_comp
_specs
= filter_component_specs
314 self
._next
_suffix
= 1
315 self
._connect
_ports
= False
317 # lists of _ComponentAndSpec
318 self
._src
_comps
_and
_specs
= []
319 self
._flt
_comps
_and
_specs
= []
323 def _compute_stream_intersections(self
):
324 # Pre-compute the trimmer range to use for each port in the graph, when
325 # stream intersection mode is enabled.
326 self
._stream
_inter
_port
_to
_range
= {}
328 for src_comp_and_spec
in self
._src
_comps
_and
_specs
:
329 # Query the port's component for the `babeltrace.trace-infos`
330 # object which contains the range for each stream, from which we can
331 # compute the intersection of the streams in each trace.
332 query_exec
= bt2
.QueryExecutor(
333 src_comp_and_spec
.spec
.component_class
,
334 'babeltrace.trace-infos',
335 src_comp_and_spec
.spec
.params
,
337 trace_infos
= query_exec
.query()
339 for trace_info
in trace_infos
:
342 stream
['range-ns']['begin']
343 for stream
in trace_info
['stream-infos']
347 [stream
['range-ns']['end'] for stream
in trace_info
['stream-infos']]
350 # Each port associated to this trace will have this computed
352 for stream
in trace_info
['stream-infos']:
353 # A port name is unique within a component, but not
354 # necessarily across all components. Use a component
355 # and port name pair to make it unique across the graph.
356 port_name
= str(stream
['port-name'])
357 key
= (src_comp_and_spec
.comp
.addr
, port_name
)
358 self
._stream
_inter
_port
_to
_range
[key
] = (begin
, end
)
360 def _validate_source_component_specs(self
, comp_specs
):
361 for comp_spec
in comp_specs
:
363 type(comp_spec
) is not ComponentSpec
364 and type(comp_spec
) is not AutoSourceComponentSpec
367 '"{}" object is not a ComponentSpec or AutoSourceComponentSpec'.format(
372 def _validate_filter_component_specs(self
, comp_specs
):
373 for comp_spec
in comp_specs
:
374 if type(comp_spec
) is not ComponentSpec
:
376 '"{}" object is not a ComponentSpec'.format(type(comp_spec
))
380 assert self
._msg
_list
[0] is None
381 self
._graph
.run_once()
382 msg
= self
._msg
_list
[0]
383 assert msg
is not None
384 self
._msg
_list
[0] = None
387 def _create_stream_intersection_trimmer(self
, component
, port
):
388 key
= (component
.addr
, port
.name
)
389 begin
, end
= self
._stream
_inter
_port
_to
_range
[key
]
390 name
= 'trimmer-{}-{}'.format(component
.name
, port
.name
)
391 return self
._create
_trimmer
(begin
, end
, name
)
393 def _create_muxer(self
):
394 plugin
= bt2
.find_plugin('utils')
397 raise RuntimeError('cannot find "utils" plugin (needed for the muxer)')
399 if 'muxer' not in plugin
.filter_component_classes
:
401 'cannot find "muxer" filter component class in "utils" plugin'
404 comp_cls
= plugin
.filter_component_classes
['muxer']
405 return self
._graph
.add_component(comp_cls
, 'muxer')
407 def _create_trimmer(self
, begin_ns
, end_ns
, name
):
408 plugin
= bt2
.find_plugin('utils')
411 raise RuntimeError('cannot find "utils" plugin (needed for the trimmer)')
413 if 'trimmer' not in plugin
.filter_component_classes
:
415 'cannot find "trimmer" filter component class in "utils" plugin'
420 def ns_to_string(ns
):
421 s_part
= ns
// 1000000000
422 ns_part
= ns
% 1000000000
423 return '{}.{:09d}'.format(s_part
, ns_part
)
425 if begin_ns
is not None:
426 params
['begin'] = ns_to_string(begin_ns
)
428 if end_ns
is not None:
429 params
['end'] = ns_to_string(end_ns
)
431 comp_cls
= plugin
.filter_component_classes
['trimmer']
432 return self
._graph
.add_component(comp_cls
, name
, params
)
434 def _get_unique_comp_name(self
, comp_cls
):
436 comps_and_specs
= itertools
.chain(
437 self
._src
_comps
_and
_specs
, self
._flt
_comps
_and
_specs
440 if name
in [comp_and_spec
.comp
.name
for comp_and_spec
in comps_and_specs
]:
441 name
+= '-{}'.format(self
._next
_suffix
)
442 self
._next
_suffix
+= 1
446 def _create_comp(self
, comp_spec
):
447 comp_cls
= comp_spec
.component_class
448 name
= self
._get
_unique
_comp
_name
(comp_cls
)
449 comp
= self
._graph
.add_component(
450 comp_cls
, name
, comp_spec
.params
, comp_spec
.obj
, comp_spec
.logging_level
454 def _get_free_muxer_input_port(self
):
455 for port
in self
._muxer
_comp
.input_ports
.values():
456 if not port
.is_connected
:
459 def _connect_src_comp_port(self
, component
, port
):
460 # if this trace collection iterator is in stream intersection
461 # mode, we need this connection:
463 # port -> trimmer -> muxer
468 if self
._stream
_intersection
_mode
:
469 trimmer_comp
= self
._create
_stream
_intersection
_trimmer
(component
, port
)
470 self
._graph
.connect_ports(port
, trimmer_comp
.input_ports
['in'])
471 port_to_muxer
= trimmer_comp
.output_ports
['out']
475 self
._graph
.connect_ports(port_to_muxer
, self
._get
_free
_muxer
_input
_port
())
477 def _graph_port_added(self
, component
, port
):
478 if not self
._connect
_ports
:
481 if type(port
) is bt2_port
._InputPortConst
:
484 if component
not in [comp
.comp
for comp
in self
._src
_comps
_and
_specs
]:
485 # do not care about non-source components (muxer, trimmer, etc.)
488 self
._connect
_src
_comp
_port
(component
, port
)
490 def _get_greatest_operative_mip_version(self
):
491 def append_comp_specs_descriptors(descriptors
, comp_specs
):
492 for comp_spec
in comp_specs
:
494 bt2
.ComponentDescriptor(
495 comp_spec
.component_class
, comp_spec
.params
, comp_spec
.obj
500 append_comp_specs_descriptors(descriptors
, self
._src
_comp
_specs
)
501 append_comp_specs_descriptors(descriptors
, self
._flt
_comp
_specs
)
503 if self
._stream
_intersection
_mode
:
504 # we also need at least one `flt.utils.trimmer` component
505 comp_spec
= ComponentSpec
.from_named_plugin_and_component_class(
508 append_comp_specs_descriptors(descriptors
, [comp_spec
])
510 mip_version
= bt2
.get_greatest_operative_mip_version(descriptors
)
512 if mip_version
is None:
513 msg
= 'failed to find an operative message interchange protocol version (components are not interoperable)'
514 raise RuntimeError(msg
)
518 def _build_graph(self
):
519 self
._graph
= bt2
.Graph(self
._get
_greatest
_operative
_mip
_version
())
520 self
._graph
.add_port_added_listener(self
._graph
_port
_added
)
521 self
._muxer
_comp
= self
._create
_muxer
()
523 if self
._begin
_ns
is not None or self
._end
_ns
is not None:
524 trimmer_comp
= self
._create
_trimmer
(self
._begin
_ns
, self
._end
_ns
, 'trimmer')
525 self
._graph
.connect_ports(
526 self
._muxer
_comp
.output_ports
['out'], trimmer_comp
.input_ports
['in']
528 last_flt_out_port
= trimmer_comp
.output_ports
['out']
530 last_flt_out_port
= self
._muxer
_comp
.output_ports
['out']
532 # create extra filter components (chained)
533 for comp_spec
in self
._flt
_comp
_specs
:
534 comp
= self
._create
_comp
(comp_spec
)
535 self
._flt
_comps
_and
_specs
.append(_ComponentAndSpec(comp
, comp_spec
))
537 # connect the extra filter chain
538 for comp_and_spec
in self
._flt
_comps
_and
_specs
:
539 in_port
= list(comp_and_spec
.comp
.input_ports
.values())[0]
540 out_port
= list(comp_and_spec
.comp
.output_ports
.values())[0]
541 self
._graph
.connect_ports(last_flt_out_port
, in_port
)
542 last_flt_out_port
= out_port
544 # Here we create the components, self._graph_port_added() is
545 # called when they add ports, but the callback returns early
546 # because self._connect_ports is False. This is because the
547 # self._graph_port_added() could not find the associated source
548 # component specification in self._src_comps_and_specs because
549 # it does not exist yet (it needs the created component to
551 for comp_spec
in self
._src
_comp
_specs
:
552 comp
= self
._create
_comp
(comp_spec
)
553 self
._src
_comps
_and
_specs
.append(_ComponentAndSpec(comp
, comp_spec
))
555 if self
._stream
_intersection
_mode
:
556 self
._compute
_stream
_intersections
()
558 # Now we connect the ports which exist at this point. We allow
559 # self._graph_port_added() to automatically connect _new_ ports.
560 self
._connect
_ports
= True
562 for comp_and_spec
in self
._src
_comps
_and
_specs
:
563 # Keep a separate list because comp_and_spec.output_ports
564 # could change during the connection of one of its ports.
565 # Any new port is handled by self._graph_port_added().
566 out_ports
= [port
for port
in comp_and_spec
.comp
.output_ports
.values()]
568 for out_port
in out_ports
:
569 if out_port
.is_connected
:
572 self
._connect
_src
_comp
_port
(comp_and_spec
.comp
, out_port
)
574 # Add the proxy sink, passing our message list to share consumed
575 # messages with this trace collection message iterator.
576 sink
= self
._graph
.add_component(
577 _TraceCollectionMessageIteratorProxySink
, 'proxy-sink', obj
=self
._msg
_list
579 sink_in_port
= sink
.input_ports
['in']
581 # connect last filter to proxy sink
582 self
._graph
.connect_ports(last_flt_out_port
, sink_in_port
)