ctf: remove unnecessary assignment
[babeltrace.git] / src / bindings / python / bt2 / bt2 / trace_collection_message_iterator.py
1 # The MIT License (MIT)
2 #
3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
4 #
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:
11 #
12 # The above copyright notice and this permission notice shall be included in
13 # all copies or substantial portions of the Software.
14 #
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
21 # THE SOFTWARE.
22
23 from bt2 import utils, native_bt
24 import bt2
25 import itertools
26 from bt2 import message_iterator as bt2_message_iterator
27 from bt2 import port as bt2_port
28 from bt2 import component as bt2_component
29 from bt2 import value as bt2_value
30 from bt2 import plugin as bt2_plugin
31 import datetime
32 from collections import namedtuple
33 import numbers
34
35
36 # a pair of component and ComponentSpec
37 _ComponentAndSpec = namedtuple('_ComponentAndSpec', ['comp', 'spec'])
38
39
40 class _BaseComponentSpec:
41 # Base for any component spec that can be passed to
42 # TraceCollectionMessageIterator.
43 def __init__(self, params, obj, logging_level):
44 if logging_level is not None:
45 utils._check_log_level(logging_level)
46
47 self._params = bt2.create_value(params)
48 self._obj = obj
49 self._logging_level = logging_level
50
51 @property
52 def params(self):
53 return self._params
54
55 @property
56 def obj(self):
57 return self._obj
58
59 @property
60 def logging_level(self):
61 return self._logging_level
62
63
64 class ComponentSpec(_BaseComponentSpec):
65 # A component spec with a specific component class.
66 def __init__(
67 self,
68 component_class,
69 params=None,
70 obj=None,
71 logging_level=bt2.LoggingLevel.NONE,
72 ):
73 if type(params) is str:
74 params = {'inputs': [params]}
75
76 super().__init__(params, obj, logging_level)
77
78 is_cc_object = isinstance(
79 component_class, (bt2._SourceComponentClass, bt2._FilterComponentClass)
80 )
81 is_user_cc_type = isinstance(
82 component_class, bt2_component._UserComponentType
83 ) and issubclass(
84 component_class, (bt2._UserSourceComponent, bt2._UserFilterComponent)
85 )
86
87 if not is_cc_object and not is_user_cc_type:
88 raise TypeError(
89 "'{}' is not a source or filter component class".format(
90 component_class.__class__.__name__
91 )
92 )
93
94 self._component_class = component_class
95
96 @property
97 def component_class(self):
98 return self._component_class
99
100 @classmethod
101 def from_named_plugin_and_component_class(
102 cls,
103 plugin_name,
104 component_class_name,
105 params=None,
106 obj=None,
107 logging_level=bt2.LoggingLevel.NONE,
108 ):
109 plugin = bt2.find_plugin(plugin_name)
110
111 if plugin is None:
112 raise ValueError('no such plugin: {}'.format(plugin_name))
113
114 if component_class_name in plugin.source_component_classes:
115 comp_class = plugin.source_component_classes[component_class_name]
116 elif component_class_name in plugin.filter_component_classes:
117 comp_class = plugin.filter_component_classes[component_class_name]
118 else:
119 raise KeyError(
120 'source or filter component class `{}` not found in plugin `{}`'.format(
121 component_class_name, plugin_name
122 )
123 )
124
125 return cls(comp_class, params, obj, logging_level)
126
127
128 class AutoSourceComponentSpec(_BaseComponentSpec):
129 # A component spec that does automatic source discovery.
130 _no_obj = object()
131
132 def __init__(self, input, params=None, obj=_no_obj, logging_level=None):
133 super().__init__(params, obj, logging_level)
134 self._input = input
135
136 @property
137 def input(self):
138 return self._input
139
140
141 def _auto_discover_source_component_specs(auto_source_comp_specs, plugin_set):
142 # Transform a list of `AutoSourceComponentSpec` in a list of `ComponentSpec`
143 # using the automatic source discovery mechanism.
144 inputs = bt2.ArrayValue([spec.input for spec in auto_source_comp_specs])
145
146 if plugin_set is None:
147 plugin_set = bt2.find_plugins()
148 else:
149 utils._check_type(plugin_set, bt2_plugin._PluginSet)
150
151 res_ptr = native_bt.bt2_auto_discover_source_components(
152 inputs._ptr, plugin_set._ptr
153 )
154
155 if res_ptr is None:
156 raise bt2._MemoryError('cannot auto discover source components')
157
158 res = bt2_value._create_from_ptr(res_ptr)
159
160 assert type(res) == bt2.MapValue
161 assert 'status' in res
162
163 status = res['status']
164 utils._handle_func_status(status, 'cannot auto-discover source components')
165
166 comp_specs = []
167 comp_specs_raw = res['results']
168 assert type(comp_specs_raw) == bt2.ArrayValue
169
170 used_input_indices = set()
171
172 for comp_spec_raw in comp_specs_raw:
173 assert type(comp_spec_raw) == bt2.ArrayValue
174 assert len(comp_spec_raw) == 4
175
176 plugin_name = comp_spec_raw[0]
177 assert type(plugin_name) == bt2.StringValue
178 plugin_name = str(plugin_name)
179
180 class_name = comp_spec_raw[1]
181 assert type(class_name) == bt2.StringValue
182 class_name = str(class_name)
183
184 comp_inputs = comp_spec_raw[2]
185 assert type(comp_inputs) == bt2.ArrayValue
186
187 comp_orig_indices = comp_spec_raw[3]
188 assert type(comp_orig_indices)
189
190 params = bt2.MapValue()
191 logging_level = bt2.LoggingLevel.NONE
192 obj = None
193
194 # Compute `params` for this component by piling up params given to all
195 # AutoSourceComponentSpec objects that contributed in the instantiation
196 # of this component.
197 #
198 # The effective log level for a component is the last one specified
199 # across the AutoSourceComponentSpec that contributed in its
200 # instantiation.
201 for idx in comp_orig_indices:
202 orig_spec = auto_source_comp_specs[idx]
203
204 if orig_spec.params is not None:
205 params.update(orig_spec.params)
206
207 if orig_spec.logging_level is not None:
208 logging_level = orig_spec.logging_level
209
210 if orig_spec.obj is not AutoSourceComponentSpec._no_obj:
211 obj = orig_spec.obj
212
213 used_input_indices.add(int(idx))
214
215 params['inputs'] = comp_inputs
216
217 comp_specs.append(
218 ComponentSpec.from_named_plugin_and_component_class(
219 plugin_name,
220 class_name,
221 params=params,
222 obj=obj,
223 logging_level=logging_level,
224 )
225 )
226
227 if len(used_input_indices) != len(inputs):
228 unused_input_indices = set(range(len(inputs))) - used_input_indices
229 unused_input_indices = sorted(unused_input_indices)
230 unused_inputs = [str(inputs[x]) for x in unused_input_indices]
231
232 msg = (
233 'Some auto source component specs did not produce any component: '
234 + ', '.join(unused_inputs)
235 )
236 raise RuntimeError(msg)
237
238 return comp_specs
239
240
241 # datetime.datetime or integral to nanoseconds
242 def _get_ns(obj):
243 if obj is None:
244 return
245
246 if isinstance(obj, numbers.Real):
247 # consider that it's already in seconds
248 s = obj
249 elif isinstance(obj, datetime.datetime):
250 # s -> ns
251 s = obj.timestamp()
252 else:
253 raise TypeError(
254 '"{}" is not an integral number or a datetime.datetime object'.format(obj)
255 )
256
257 return int(s * 1e9)
258
259
260 class _TraceCollectionMessageIteratorProxySink(bt2_component._UserSinkComponent):
261 def __init__(self, params, msg_list):
262 assert type(msg_list) is list
263 self._msg_list = msg_list
264 self._add_input_port('in')
265
266 def _user_graph_is_configured(self):
267 self._msg_iter = self._create_input_port_message_iterator(
268 self._input_ports['in']
269 )
270
271 def _user_consume(self):
272 assert self._msg_list[0] is None
273 self._msg_list[0] = next(self._msg_iter)
274
275
276 class TraceCollectionMessageIterator(bt2_message_iterator._MessageIterator):
277 def __init__(
278 self,
279 source_component_specs,
280 filter_component_specs=None,
281 stream_intersection_mode=False,
282 begin=None,
283 end=None,
284 plugin_set=None,
285 ):
286 utils._check_bool(stream_intersection_mode)
287 self._stream_intersection_mode = stream_intersection_mode
288 self._begin_ns = _get_ns(begin)
289 self._end_ns = _get_ns(end)
290 self._msg_list = [None]
291
292 # If a single item is provided, convert to a list.
293 if type(source_component_specs) in (
294 ComponentSpec,
295 AutoSourceComponentSpec,
296 str,
297 ):
298 source_component_specs = [source_component_specs]
299
300 # Convert any string to an AutoSourceComponentSpec.
301 def str_to_auto(item):
302 if type(item) is str:
303 item = AutoSourceComponentSpec(item)
304
305 return item
306
307 source_component_specs = [str_to_auto(s) for s in source_component_specs]
308
309 if type(filter_component_specs) is ComponentSpec:
310 filter_component_specs = [filter_component_specs]
311 elif filter_component_specs is None:
312 filter_component_specs = []
313
314 self._validate_source_component_specs(source_component_specs)
315 self._validate_filter_component_specs(filter_component_specs)
316
317 # Pass any `ComponentSpec` instance as-is.
318 self._src_comp_specs = [
319 spec for spec in source_component_specs if type(spec) is ComponentSpec
320 ]
321
322 # Convert any `AutoSourceComponentSpec` in concrete `ComponentSpec` instances.
323 auto_src_comp_specs = [
324 spec
325 for spec in source_component_specs
326 if type(spec) is AutoSourceComponentSpec
327 ]
328 self._src_comp_specs += _auto_discover_source_component_specs(
329 auto_src_comp_specs, plugin_set
330 )
331
332 self._flt_comp_specs = filter_component_specs
333 self._next_suffix = 1
334 self._connect_ports = False
335
336 # lists of _ComponentAndSpec
337 self._src_comps_and_specs = []
338 self._flt_comps_and_specs = []
339
340 self._build_graph()
341
342 def _validate_source_component_specs(self, comp_specs):
343 for comp_spec in comp_specs:
344 if (
345 type(comp_spec) is not ComponentSpec
346 and type(comp_spec) is not AutoSourceComponentSpec
347 ):
348 raise TypeError(
349 '"{}" object is not a ComponentSpec or AutoSourceComponentSpec'.format(
350 type(comp_spec)
351 )
352 )
353
354 def _validate_filter_component_specs(self, comp_specs):
355 for comp_spec in comp_specs:
356 if type(comp_spec) is not ComponentSpec:
357 raise TypeError(
358 '"{}" object is not a ComponentSpec'.format(type(comp_spec))
359 )
360
361 def __next__(self):
362 assert self._msg_list[0] is None
363 self._graph.run_once()
364 msg = self._msg_list[0]
365 assert msg is not None
366 self._msg_list[0] = None
367 return msg
368
369 def _create_stream_intersection_trimmer(self, component, port):
370 # find the original parameters specified by the user to create
371 # this port's component to get the `inputs` parameter
372 for src_comp_and_spec in self._src_comps_and_specs:
373 if component == src_comp_and_spec.comp:
374 break
375
376 try:
377 inputs = src_comp_and_spec.spec.params['inputs']
378 except Exception as e:
379 raise ValueError(
380 'all source components must be created with an "inputs" parameter in stream intersection mode'
381 ) from e
382
383 params = {'inputs': inputs}
384
385 # query the port's component for the `babeltrace.trace-info`
386 # object which contains the stream intersection range for each
387 # exposed trace
388 query_exec = bt2.QueryExecutor(
389 src_comp_and_spec.comp.cls, 'babeltrace.trace-info', params
390 )
391 trace_info_res = query_exec.query()
392 begin = None
393 end = None
394
395 # find the trace info for this port's trace
396 try:
397 for trace_info in trace_info_res:
398 for stream in trace_info['streams']:
399 if stream['port-name'] == port.name:
400 range_ns = trace_info['intersection-range-ns']
401 begin = range_ns['begin']
402 end = range_ns['end']
403 break
404 except Exception:
405 pass
406
407 if begin is None or end is None:
408 raise RuntimeError(
409 'cannot find stream intersection range for port "{}"'.format(port.name)
410 )
411
412 name = 'trimmer-{}-{}'.format(src_comp_and_spec.comp.name, port.name)
413 return self._create_trimmer(begin, end, name)
414
415 def _create_muxer(self):
416 plugin = bt2.find_plugin('utils')
417
418 if plugin is None:
419 raise RuntimeError('cannot find "utils" plugin (needed for the muxer)')
420
421 if 'muxer' not in plugin.filter_component_classes:
422 raise RuntimeError(
423 'cannot find "muxer" filter component class in "utils" plugin'
424 )
425
426 comp_cls = plugin.filter_component_classes['muxer']
427 return self._graph.add_component(comp_cls, 'muxer')
428
429 def _create_trimmer(self, begin_ns, end_ns, name):
430 plugin = bt2.find_plugin('utils')
431
432 if plugin is None:
433 raise RuntimeError('cannot find "utils" plugin (needed for the trimmer)')
434
435 if 'trimmer' not in plugin.filter_component_classes:
436 raise RuntimeError(
437 'cannot find "trimmer" filter component class in "utils" plugin'
438 )
439
440 params = {}
441
442 def ns_to_string(ns):
443 s_part = ns // 1000000000
444 ns_part = ns % 1000000000
445 return '{}.{:09d}'.format(s_part, ns_part)
446
447 if begin_ns is not None:
448 params['begin'] = ns_to_string(begin_ns)
449
450 if end_ns is not None:
451 params['end'] = ns_to_string(end_ns)
452
453 comp_cls = plugin.filter_component_classes['trimmer']
454 return self._graph.add_component(comp_cls, name, params)
455
456 def _get_unique_comp_name(self, comp_cls):
457 name = comp_cls.name
458 comps_and_specs = itertools.chain(
459 self._src_comps_and_specs, self._flt_comps_and_specs
460 )
461
462 if name in [comp_and_spec.comp.name for comp_and_spec in comps_and_specs]:
463 name += '-{}'.format(self._next_suffix)
464 self._next_suffix += 1
465
466 return name
467
468 def _create_comp(self, comp_spec):
469 comp_cls = comp_spec.component_class
470 name = self._get_unique_comp_name(comp_cls)
471 comp = self._graph.add_component(
472 comp_cls, name, comp_spec.params, comp_spec.obj, comp_spec.logging_level
473 )
474 return comp
475
476 def _get_free_muxer_input_port(self):
477 for port in self._muxer_comp.input_ports.values():
478 if not port.is_connected:
479 return port
480
481 def _connect_src_comp_port(self, component, port):
482 # if this trace collection iterator is in stream intersection
483 # mode, we need this connection:
484 #
485 # port -> trimmer -> muxer
486 #
487 # otherwise, simply:
488 #
489 # port -> muxer
490 if self._stream_intersection_mode:
491 trimmer_comp = self._create_stream_intersection_trimmer(component, port)
492 self._graph.connect_ports(port, trimmer_comp.input_ports['in'])
493 port_to_muxer = trimmer_comp.output_ports['out']
494 else:
495 port_to_muxer = port
496
497 self._graph.connect_ports(port_to_muxer, self._get_free_muxer_input_port())
498
499 def _graph_port_added(self, component, port):
500 if not self._connect_ports:
501 return
502
503 if type(port) is bt2_port._InputPort:
504 return
505
506 if component not in [comp.comp for comp in self._src_comps_and_specs]:
507 # do not care about non-source components (muxer, trimmer, etc.)
508 return
509
510 self._connect_src_comp_port(component, port)
511
512 def _get_greatest_operative_mip_version(self):
513 def append_comp_specs_descriptors(descriptors, comp_specs):
514 for comp_spec in comp_specs:
515 descriptors.append(
516 bt2.ComponentDescriptor(
517 comp_spec.component_class, comp_spec.params, comp_spec.obj
518 )
519 )
520
521 descriptors = []
522 append_comp_specs_descriptors(descriptors, self._src_comp_specs)
523 append_comp_specs_descriptors(descriptors, self._flt_comp_specs)
524
525 if self._stream_intersection_mode:
526 # we also need at least one `flt.utils.trimmer` component
527 comp_spec = ComponentSpec.from_named_plugin_and_component_class(
528 'utils', 'trimmer'
529 )
530 append_comp_specs_descriptors(descriptors, [comp_spec])
531
532 mip_version = bt2.get_greatest_operative_mip_version(descriptors)
533
534 if mip_version is None:
535 msg = 'failed to find an operative message interchange protocol version (components are not interoperable)'
536 raise RuntimeError(msg)
537
538 return mip_version
539
540 def _build_graph(self):
541 self._graph = bt2.Graph(self._get_greatest_operative_mip_version())
542 self._graph.add_port_added_listener(self._graph_port_added)
543 self._muxer_comp = self._create_muxer()
544
545 if self._begin_ns is not None or self._end_ns is not None:
546 trimmer_comp = self._create_trimmer(self._begin_ns, self._end_ns, 'trimmer')
547 self._graph.connect_ports(
548 self._muxer_comp.output_ports['out'], trimmer_comp.input_ports['in']
549 )
550 last_flt_out_port = trimmer_comp.output_ports['out']
551 else:
552 last_flt_out_port = self._muxer_comp.output_ports['out']
553
554 # create extra filter components (chained)
555 for comp_spec in self._flt_comp_specs:
556 comp = self._create_comp(comp_spec)
557 self._flt_comps_and_specs.append(_ComponentAndSpec(comp, comp_spec))
558
559 # connect the extra filter chain
560 for comp_and_spec in self._flt_comps_and_specs:
561 in_port = list(comp_and_spec.comp.input_ports.values())[0]
562 out_port = list(comp_and_spec.comp.output_ports.values())[0]
563 self._graph.connect_ports(last_flt_out_port, in_port)
564 last_flt_out_port = out_port
565
566 # Here we create the components, self._graph_port_added() is
567 # called when they add ports, but the callback returns early
568 # because self._connect_ports is False. This is because the
569 # self._graph_port_added() could not find the associated source
570 # component specification in self._src_comps_and_specs because
571 # it does not exist yet (it needs the created component to
572 # exist).
573 for comp_spec in self._src_comp_specs:
574 comp = self._create_comp(comp_spec)
575 self._src_comps_and_specs.append(_ComponentAndSpec(comp, comp_spec))
576
577 # Now we connect the ports which exist at this point. We allow
578 # self._graph_port_added() to automatically connect _new_ ports.
579 self._connect_ports = True
580
581 for comp_and_spec in self._src_comps_and_specs:
582 # Keep a separate list because comp_and_spec.output_ports
583 # could change during the connection of one of its ports.
584 # Any new port is handled by self._graph_port_added().
585 out_ports = [port for port in comp_and_spec.comp.output_ports.values()]
586
587 for out_port in out_ports:
588 if out_port.is_connected:
589 continue
590
591 self._connect_src_comp_port(comp_and_spec.comp, out_port)
592
593 # Add the proxy sink, passing our message list to share consumed
594 # messages with this trace collection message iterator.
595 sink = self._graph.add_component(
596 _TraceCollectionMessageIteratorProxySink, 'proxy-sink', obj=self._msg_list
597 )
598 sink_in_port = sink.input_ports['in']
599
600 # connect last filter to proxy sink
601 self._graph.connect_ports(last_flt_out_port, sink_in_port)
This page took 0.042969 seconds and 5 git commands to generate.