1 # SPDX-License-Identifier: MIT
3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
8 from collections
import namedtuple
10 from bt2
import mip
as bt2_mip
11 from bt2
import port
as bt2_port
12 from bt2
import error
as bt2_error
13 from bt2
import graph
as bt2_graph
14 from bt2
import utils
as bt2_utils
15 from bt2
import value
as bt2_value
16 from bt2
import plugin
as bt2_plugin
17 from bt2
import logging
as bt2_logging
18 from bt2
import component
as bt2_component
19 from bt2
import native_bt
20 from bt2
import query_executor
as bt2_query_executor
21 from bt2
import message_iterator
as bt2_message_iterator
22 from bt2
import component_descriptor
as bt2_component_descriptor
24 # a pair of component and ComponentSpec
25 _ComponentAndSpec
= namedtuple("_ComponentAndSpec", ["comp", "spec"])
28 class _BaseComponentSpec
:
29 # Base for any component spec that can be passed to
30 # TraceCollectionMessageIterator.
31 def __init__(self
, params
, obj
, logging_level
):
32 if logging_level
is not None:
33 bt2_utils
._check
_log
_level
(logging_level
)
35 self
._params
= bt2_value
.create_value(params
)
37 self
._logging
_level
= logging_level
48 def logging_level(self
):
49 return self
._logging
_level
52 class ComponentSpec(_BaseComponentSpec
):
53 # A component spec with a specific component class.
59 logging_level
=bt2_logging
.LoggingLevel
.NONE
,
61 if type(params
) is str:
62 params
= {"inputs": [params
]}
64 super().__init
__(params
, obj
, logging_level
)
66 is_cc_object
= isinstance(
69 bt2_component
._SourceComponentClassConst
,
70 bt2_component
._FilterComponentClassConst
,
73 is_user_cc_type
= isinstance(
74 component_class
, bt2_component
._UserComponentType
77 (bt2_component
._UserSourceComponent
, bt2_component
._UserFilterComponent
),
80 if not is_cc_object
and not is_user_cc_type
:
82 "'{}' is not a source or filter component class".format(
83 component_class
.__class
__.__name
__
87 self
._component
_class
= component_class
90 def component_class(self
):
91 return self
._component
_class
94 def from_named_plugin_and_component_class(
100 logging_level
=bt2_logging
.LoggingLevel
.NONE
,
102 plugin
= bt2_plugin
.find_plugin(plugin_name
)
105 raise ValueError("no such plugin: {}".format(plugin_name
))
107 if component_class_name
in plugin
.source_component_classes
:
108 comp_class
= plugin
.source_component_classes
[component_class_name
]
109 elif component_class_name
in plugin
.filter_component_classes
:
110 comp_class
= plugin
.filter_component_classes
[component_class_name
]
113 "source or filter component class `{}` not found in plugin `{}`".format(
114 component_class_name
, plugin_name
118 return cls(comp_class
, params
, obj
, logging_level
)
121 class AutoSourceComponentSpec(_BaseComponentSpec
):
122 # A component spec that does automatic source discovery.
125 def __init__(self
, input, params
=None, obj
=_no_obj
, logging_level
=None):
126 super().__init
__(params
, obj
, logging_level
)
134 def _auto_discover_source_component_specs(auto_source_comp_specs
, plugin_set
):
135 # Transform a list of `AutoSourceComponentSpec` in a list of `ComponentSpec`
136 # using the automatic source discovery mechanism.
137 inputs
= bt2_value
.ArrayValue([spec
.input for spec
in auto_source_comp_specs
])
139 if plugin_set
is None:
140 plugin_set
= bt2_plugin
.find_plugins()
142 bt2_utils
._check
_type
(plugin_set
, bt2_plugin
._PluginSet
)
144 res_ptr
= native_bt
.bt2_auto_discover_source_components(
145 inputs
._ptr
, plugin_set
._ptr
149 raise bt2_error
._MemoryError
("cannot auto discover source components")
151 res
= bt2_value
._create
_from
_ptr
(res_ptr
)
153 assert type(res
) is bt2_value
.MapValue
154 assert "status" in res
156 status
= res
["status"]
157 bt2_utils
._handle
_func
_status
(status
, "cannot auto-discover source components")
160 comp_specs_raw
= res
["results"]
161 assert type(comp_specs_raw
) is bt2_value
.ArrayValue
163 used_input_indices
= set()
165 for comp_spec_raw
in comp_specs_raw
:
166 assert type(comp_spec_raw
) is bt2_value
.ArrayValue
167 assert len(comp_spec_raw
) == 4
169 plugin_name
= comp_spec_raw
[0]
170 assert type(plugin_name
) is bt2_value
.StringValue
171 plugin_name
= str(plugin_name
)
173 class_name
= comp_spec_raw
[1]
174 assert type(class_name
) is bt2_value
.StringValue
175 class_name
= str(class_name
)
177 comp_inputs
= comp_spec_raw
[2]
178 assert type(comp_inputs
) is bt2_value
.ArrayValue
180 comp_orig_indices
= comp_spec_raw
[3]
181 assert type(comp_orig_indices
)
183 params
= bt2_value
.MapValue()
184 logging_level
= bt2_logging
.LoggingLevel
.NONE
187 # Compute `params` for this component by piling up params given to all
188 # AutoSourceComponentSpec objects that contributed in the instantiation
191 # The effective log level for a component is the last one specified
192 # across the AutoSourceComponentSpec that contributed in its
194 for idx
in comp_orig_indices
:
195 orig_spec
= auto_source_comp_specs
[idx
]
197 if orig_spec
.params
is not None:
198 params
.update(orig_spec
.params
)
200 if orig_spec
.logging_level
is not None:
201 logging_level
= orig_spec
.logging_level
203 if orig_spec
.obj
is not AutoSourceComponentSpec
._no
_obj
:
206 used_input_indices
.add(int(idx
))
208 params
["inputs"] = comp_inputs
211 ComponentSpec
.from_named_plugin_and_component_class(
216 logging_level
=logging_level
,
220 if len(used_input_indices
) != len(inputs
):
221 unused_input_indices
= set(range(len(inputs
))) - used_input_indices
222 unused_input_indices
= sorted(unused_input_indices
)
223 unused_inputs
= [str(inputs
[x
]) for x
in unused_input_indices
]
226 "Some auto source component specs did not produce any component: "
227 + ", ".join(unused_inputs
)
229 raise RuntimeError(msg
)
234 # datetime.datetime or integral to nanoseconds
239 if isinstance(obj
, numbers
.Real
):
240 # consider that it's already in seconds
242 elif isinstance(obj
, datetime
.datetime
):
247 '"{}" is not an integral number or a datetime.datetime object'.format(obj
)
253 class _TraceCollectionMessageIteratorProxySink(bt2_component
._UserSinkComponent
):
254 def __init__(self
, config
, params
, msg_list
):
255 assert type(msg_list
) is list
256 self
._msg
_list
= msg_list
257 self
._add
_input
_port
("in")
259 def _user_graph_is_configured(self
):
260 self
._msg
_iter
= self
._create
_message
_iterator
(self
._input
_ports
["in"])
262 def _user_consume(self
):
263 assert self
._msg
_list
[0] is None
264 self
._msg
_list
[0] = next(self
._msg
_iter
)
267 class TraceCollectionMessageIterator(bt2_message_iterator
._MessageIterator
):
270 source_component_specs
,
271 filter_component_specs
=None,
272 stream_intersection_mode
=False,
277 bt2_utils
._check
_bool
(stream_intersection_mode
)
278 self
._stream
_intersection
_mode
= stream_intersection_mode
279 self
._begin
_ns
= _get_ns(begin
)
280 self
._end
_ns
= _get_ns(end
)
281 self
._msg
_list
= [None]
283 # If a single item is provided, convert to a list.
284 if type(source_component_specs
) in (
286 AutoSourceComponentSpec
,
289 source_component_specs
= [source_component_specs
]
291 # Convert any string to an AutoSourceComponentSpec.
292 def str_to_auto(item
):
293 if type(item
) is str:
294 item
= AutoSourceComponentSpec(item
)
298 source_component_specs
= [str_to_auto(s
) for s
in source_component_specs
]
300 if type(filter_component_specs
) is ComponentSpec
:
301 filter_component_specs
= [filter_component_specs
]
302 elif filter_component_specs
is None:
303 filter_component_specs
= []
305 self
._validate
_source
_component
_specs
(source_component_specs
)
306 self
._validate
_filter
_component
_specs
(filter_component_specs
)
308 # Pass any `ComponentSpec` instance as-is.
309 self
._src
_comp
_specs
= [
310 spec
for spec
in source_component_specs
if type(spec
) is ComponentSpec
313 # Convert any `AutoSourceComponentSpec` in concrete `ComponentSpec` instances.
314 auto_src_comp_specs
= [
316 for spec
in source_component_specs
317 if type(spec
) is AutoSourceComponentSpec
319 self
._src
_comp
_specs
+= _auto_discover_source_component_specs(
320 auto_src_comp_specs
, plugin_set
323 self
._flt
_comp
_specs
= filter_component_specs
324 self
._next
_suffix
= 1
325 self
._connect
_ports
= False
327 # lists of _ComponentAndSpec
328 self
._src
_comps
_and
_specs
= []
329 self
._flt
_comps
_and
_specs
= []
333 def _compute_stream_intersections(self
):
334 # Pre-compute the trimmer range to use for each port in the graph, when
335 # stream intersection mode is enabled.
336 self
._stream
_inter
_port
_to
_range
= {}
338 for src_comp_and_spec
in self
._src
_comps
_and
_specs
:
339 # Query the port's component for the `babeltrace.trace-infos`
340 # object which contains the range for each stream, from which we can
341 # compute the intersection of the streams in each trace.
342 query_exec
= bt2_query_executor
.QueryExecutor(
343 src_comp_and_spec
.spec
.component_class
,
344 "babeltrace.trace-infos",
345 src_comp_and_spec
.spec
.params
,
347 trace_infos
= query_exec
.query()
349 for trace_info
in trace_infos
:
352 stream
["range-ns"]["begin"]
353 for stream
in trace_info
["stream-infos"]
357 [stream
["range-ns"]["end"] for stream
in trace_info
["stream-infos"]]
360 # Each port associated to this trace will have this computed
362 for stream
in trace_info
["stream-infos"]:
363 # A port name is unique within a component, but not
364 # necessarily across all components. Use a component
365 # and port name pair to make it unique across the graph.
366 port_name
= str(stream
["port-name"])
367 key
= (src_comp_and_spec
.comp
.addr
, port_name
)
368 self
._stream
_inter
_port
_to
_range
[key
] = (begin
, end
)
370 def _validate_source_component_specs(self
, comp_specs
):
371 for comp_spec
in comp_specs
:
373 type(comp_spec
) is not ComponentSpec
374 and type(comp_spec
) is not AutoSourceComponentSpec
377 '"{}" object is not a ComponentSpec or AutoSourceComponentSpec'.format(
382 def _validate_filter_component_specs(self
, comp_specs
):
383 for comp_spec
in comp_specs
:
384 if type(comp_spec
) is not ComponentSpec
:
386 '"{}" object is not a ComponentSpec'.format(type(comp_spec
))
390 assert self
._msg
_list
[0] is None
391 self
._graph
.run_once()
392 msg
= self
._msg
_list
[0]
393 assert msg
is not None
394 self
._msg
_list
[0] = None
397 def _create_stream_intersection_trimmer(self
, component
, port
):
398 key
= (component
.addr
, port
.name
)
399 begin
, end
= self
._stream
_inter
_port
_to
_range
[key
]
400 name
= "trimmer-{}-{}".format(component
.name
, port
.name
)
401 return self
._create
_trimmer
(begin
, end
, name
)
403 def _create_muxer(self
):
404 plugin
= bt2_plugin
.find_plugin("utils")
407 raise RuntimeError('cannot find "utils" plugin (needed for the muxer)')
409 if "muxer" not in plugin
.filter_component_classes
:
411 'cannot find "muxer" filter component class in "utils" plugin'
414 comp_cls
= plugin
.filter_component_classes
["muxer"]
415 return self
._graph
.add_component(comp_cls
, "muxer")
417 def _create_trimmer(self
, begin_ns
, end_ns
, name
):
418 plugin
= bt2_plugin
.find_plugin("utils")
421 raise RuntimeError('cannot find "utils" plugin (needed for the trimmer)')
423 if "trimmer" not in plugin
.filter_component_classes
:
425 'cannot find "trimmer" filter component class in "utils" plugin'
430 def ns_to_string(ns
):
431 s_part
= ns
// 1000000000
432 ns_part
= ns
% 1000000000
433 return "{}.{:09d}".format(s_part
, ns_part
)
435 if begin_ns
is not None:
436 params
["begin"] = ns_to_string(begin_ns
)
438 if end_ns
is not None:
439 params
["end"] = ns_to_string(end_ns
)
441 comp_cls
= plugin
.filter_component_classes
["trimmer"]
442 return self
._graph
.add_component(comp_cls
, name
, params
)
444 def _get_unique_comp_name(self
, comp_cls
):
446 comps_and_specs
= itertools
.chain(
447 self
._src
_comps
_and
_specs
, self
._flt
_comps
_and
_specs
450 if name
in [comp_and_spec
.comp
.name
for comp_and_spec
in comps_and_specs
]:
451 name
+= "-{}".format(self
._next
_suffix
)
452 self
._next
_suffix
+= 1
456 def _create_comp(self
, comp_spec
):
457 comp_cls
= comp_spec
.component_class
458 name
= self
._get
_unique
_comp
_name
(comp_cls
)
459 comp
= self
._graph
.add_component(
460 comp_cls
, name
, comp_spec
.params
, comp_spec
.obj
, comp_spec
.logging_level
464 def _get_free_muxer_input_port(self
):
465 for port
in self
._muxer
_comp
.input_ports
.values():
466 if not port
.is_connected
:
469 def _connect_src_comp_port(self
, component
, port
):
470 # if this trace collection iterator is in stream intersection
471 # mode, we need this connection:
473 # port -> trimmer -> muxer
478 if self
._stream
_intersection
_mode
:
479 trimmer_comp
= self
._create
_stream
_intersection
_trimmer
(component
, port
)
480 self
._graph
.connect_ports(port
, trimmer_comp
.input_ports
["in"])
481 port_to_muxer
= trimmer_comp
.output_ports
["out"]
485 self
._graph
.connect_ports(port_to_muxer
, self
._get
_free
_muxer
_input
_port
())
487 def _graph_port_added(self
, component
, port
):
488 if not self
._connect
_ports
:
491 if type(port
) is bt2_port
._InputPortConst
:
494 if component
not in [comp
.comp
for comp
in self
._src
_comps
_and
_specs
]:
495 # do not care about non-source components (muxer, trimmer, etc.)
498 self
._connect
_src
_comp
_port
(component
, port
)
500 def _get_greatest_operative_mip_version(self
):
501 def append_comp_specs_descriptors(descriptors
, comp_specs
):
502 for comp_spec
in comp_specs
:
504 bt2_component_descriptor
.ComponentDescriptor(
505 comp_spec
.component_class
, comp_spec
.params
, comp_spec
.obj
510 append_comp_specs_descriptors(descriptors
, self
._src
_comp
_specs
)
511 append_comp_specs_descriptors(descriptors
, self
._flt
_comp
_specs
)
513 if self
._stream
_intersection
_mode
:
514 # we also need at least one `flt.utils.trimmer` component
515 comp_spec
= ComponentSpec
.from_named_plugin_and_component_class(
518 append_comp_specs_descriptors(descriptors
, [comp_spec
])
520 mip_version
= bt2_mip
.get_greatest_operative_mip_version(descriptors
)
522 if mip_version
is None:
523 msg
= "failed to find an operative message interchange protocol version (components are not interoperable)"
524 raise RuntimeError(msg
)
528 def _build_graph(self
):
529 self
._graph
= bt2_graph
.Graph(self
._get
_greatest
_operative
_mip
_version
())
530 self
._graph
.add_port_added_listener(self
._graph
_port
_added
)
531 self
._muxer
_comp
= self
._create
_muxer
()
533 if self
._begin
_ns
is not None or self
._end
_ns
is not None:
534 trimmer_comp
= self
._create
_trimmer
(self
._begin
_ns
, self
._end
_ns
, "trimmer")
535 self
._graph
.connect_ports(
536 self
._muxer
_comp
.output_ports
["out"], trimmer_comp
.input_ports
["in"]
538 last_flt_out_port
= trimmer_comp
.output_ports
["out"]
540 last_flt_out_port
= self
._muxer
_comp
.output_ports
["out"]
542 # create extra filter components (chained)
543 for comp_spec
in self
._flt
_comp
_specs
:
544 comp
= self
._create
_comp
(comp_spec
)
545 self
._flt
_comps
_and
_specs
.append(_ComponentAndSpec(comp
, comp_spec
))
547 # connect the extra filter chain
548 for comp_and_spec
in self
._flt
_comps
_and
_specs
:
549 in_port
= list(comp_and_spec
.comp
.input_ports
.values())[0]
550 out_port
= list(comp_and_spec
.comp
.output_ports
.values())[0]
551 self
._graph
.connect_ports(last_flt_out_port
, in_port
)
552 last_flt_out_port
= out_port
554 # Here we create the components, self._graph_port_added() is
555 # called when they add ports, but the callback returns early
556 # because self._connect_ports is False. This is because the
557 # self._graph_port_added() could not find the associated source
558 # component specification in self._src_comps_and_specs because
559 # it does not exist yet (it needs the created component to
561 for comp_spec
in self
._src
_comp
_specs
:
562 comp
= self
._create
_comp
(comp_spec
)
563 self
._src
_comps
_and
_specs
.append(_ComponentAndSpec(comp
, comp_spec
))
565 if self
._stream
_intersection
_mode
:
566 self
._compute
_stream
_intersections
()
568 # Now we connect the ports which exist at this point. We allow
569 # self._graph_port_added() to automatically connect _new_ ports.
570 self
._connect
_ports
= True
572 for comp_and_spec
in self
._src
_comps
_and
_specs
:
573 # Keep a separate list because comp_and_spec.output_ports
574 # could change during the connection of one of its ports.
575 # Any new port is handled by self._graph_port_added().
576 out_ports
= [port
for port
in comp_and_spec
.comp
.output_ports
.values()]
578 for out_port
in out_ports
:
579 if out_port
.is_connected
:
582 self
._connect
_src
_comp
_port
(comp_and_spec
.comp
, out_port
)
584 # Add the proxy sink, passing our message list to share consumed
585 # messages with this trace collection message iterator.
586 sink
= self
._graph
.add_component(
587 _TraceCollectionMessageIteratorProxySink
, "proxy-sink", obj
=self
._msg
_list
589 sink_in_port
= sink
.input_ports
["in"]
591 # connect last filter to proxy sink
592 self
._graph
.connect_ports(last_flt_out_port
, sink_in_port
)