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
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
31 from collections
import namedtuple
35 # a pair of component and ComponentSpec
36 _ComponentAndSpec
= namedtuple('_ComponentAndSpec', ['comp', 'spec'])
46 logging_level
=bt2_logging
.LoggingLevel
.NONE
,
48 utils
._check
_str
(plugin_name
)
49 utils
._check
_str
(class_name
)
50 utils
._check
_log
_level
(logging_level
)
51 self
._plugin
_name
= plugin_name
52 self
._class
_name
= class_name
53 self
._logging
_level
= logging_level
56 if type(params
) is str:
57 self
._params
= bt2
.create_value({'inputs': [params
]})
59 self
._params
= bt2
.create_value(params
)
62 def plugin_name(self
):
63 return self
._plugin
_name
67 return self
._class
_name
70 def logging_level(self
):
71 return self
._logging
_level
82 # datetime.datetime or integral to nanoseconds
87 if isinstance(obj
, numbers
.Real
):
88 # consider that it's already in seconds
90 elif isinstance(obj
, datetime
.datetime
):
95 '"{}" is not an integral number or a datetime.datetime object'.format(obj
)
106 class _TraceCollectionMessageIteratorProxySink(bt2_component
._UserSinkComponent
):
107 def __init__(self
, params
, msg_list
):
108 assert type(msg_list
) is list
109 self
._msg
_list
= msg_list
110 self
._add
_input
_port
('in')
112 def _user_graph_is_configured(self
):
113 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
114 self
._input
_ports
['in']
117 def _user_consume(self
):
118 assert self
._msg
_list
[0] is None
119 self
._msg
_list
[0] = next(self
._msg
_iter
)
122 class TraceCollectionMessageIterator(bt2_message_iterator
._MessageIterator
):
125 source_component_specs
,
126 filter_component_specs
=None,
127 stream_intersection_mode
=False,
131 utils
._check
_bool
(stream_intersection_mode
)
132 self
._stream
_intersection
_mode
= stream_intersection_mode
133 self
._begin
_ns
= _get_ns(begin
)
134 self
._end
_ns
= _get_ns(end
)
135 self
._msg
_list
= [None]
137 if type(source_component_specs
) is ComponentSpec
:
138 source_component_specs
= [source_component_specs
]
140 if type(filter_component_specs
) is ComponentSpec
:
141 filter_component_specs
= [filter_component_specs
]
142 elif filter_component_specs
is None:
143 filter_component_specs
= []
145 self
._src
_comp
_specs
= source_component_specs
146 self
._flt
_comp
_specs
= filter_component_specs
147 self
._next
_suffix
= 1
148 self
._connect
_ports
= False
150 # lists of _ComponentAndSpec
151 self
._src
_comps
_and
_specs
= []
152 self
._flt
_comps
_and
_specs
= []
154 self
._validate
_component
_specs
(source_component_specs
)
155 self
._validate
_component
_specs
(filter_component_specs
)
158 def _validate_component_specs(self
, comp_specs
):
159 for comp_spec
in comp_specs
:
160 if type(comp_spec
) is not ComponentSpec
:
162 '"{}" object is not a ComponentSpec'.format(type(comp_spec
))
166 assert self
._msg
_list
[0] is None
167 self
._graph
.run_once()
168 msg
= self
._msg
_list
[0]
169 assert msg
is not None
170 self
._msg
_list
[0] = None
173 def _create_stream_intersection_trimmer(self
, component
, port
):
174 # find the original parameters specified by the user to create
175 # this port's component to get the `inputs` parameter
176 for src_comp_and_spec
in self
._src
_comps
_and
_specs
:
177 if component
== src_comp_and_spec
.comp
:
181 inputs
= src_comp_and_spec
.spec
.params
['inputs']
182 except Exception as e
:
184 'all source components must be created with an "inputs" parameter in stream intersection mode'
187 params
= {'inputs': inputs
}
189 # query the port's component for the `babeltrace.trace-info`
190 # object which contains the stream intersection range for each
192 query_exec
= bt2
.QueryExecutor(
193 src_comp_and_spec
.comp
.cls
, 'babeltrace.trace-info', params
195 trace_info_res
= query_exec
.query()
199 # find the trace info for this port's trace
201 for trace_info
in trace_info_res
:
202 for stream
in trace_info
['streams']:
203 if stream
['port-name'] == port
.name
:
204 range_ns
= trace_info
['intersection-range-ns']
205 begin
= range_ns
['begin']
206 end
= range_ns
['end']
211 if begin
is None or end
is None:
213 'cannot find stream intersection range for port "{}"'.format(port
.name
)
216 name
= 'trimmer-{}-{}'.format(src_comp_and_spec
.comp
.name
, port
.name
)
217 return self
._create
_trimmer
(begin
, end
, name
)
219 def _create_muxer(self
):
220 plugin
= bt2
.find_plugin('utils')
223 raise RuntimeError('cannot find "utils" plugin (needed for the muxer)')
225 if 'muxer' not in plugin
.filter_component_classes
:
227 'cannot find "muxer" filter component class in "utils" plugin'
230 comp_cls
= plugin
.filter_component_classes
['muxer']
231 return self
._graph
.add_component(comp_cls
, 'muxer')
233 def _create_trimmer(self
, begin_ns
, end_ns
, name
):
234 plugin
= bt2
.find_plugin('utils')
237 raise RuntimeError('cannot find "utils" plugin (needed for the trimmer)')
239 if 'trimmer' not in plugin
.filter_component_classes
:
241 'cannot find "trimmer" filter component class in "utils" plugin'
246 def ns_to_string(ns
):
247 s_part
= ns
// 1000000000
248 ns_part
= ns
% 1000000000
249 return '{}.{:09d}'.format(s_part
, ns_part
)
251 if begin_ns
is not None:
252 params
['begin'] = ns_to_string(begin_ns
)
254 if end_ns
is not None:
255 params
['end'] = ns_to_string(end_ns
)
257 comp_cls
= plugin
.filter_component_classes
['trimmer']
258 return self
._graph
.add_component(comp_cls
, name
, params
)
260 def _get_unique_comp_name(self
, comp_spec
):
261 name
= '{}-{}'.format(comp_spec
.plugin_name
, comp_spec
.class_name
)
262 comps_and_specs
= itertools
.chain(
263 self
._src
_comps
_and
_specs
, self
._flt
_comps
_and
_specs
266 if name
in [comp_and_spec
.comp
.name
for comp_and_spec
in comps_and_specs
]:
267 name
+= '-{}'.format(self
._next
_suffix
)
268 self
._next
_suffix
+= 1
272 def _create_comp(self
, comp_spec
, comp_cls_type
):
273 plugin
= bt2
.find_plugin(comp_spec
.plugin_name
)
276 raise ValueError('no such plugin: {}'.format(comp_spec
.plugin_name
))
278 if comp_cls_type
== _CompClsType
.SOURCE
:
279 comp_classes
= plugin
.source_component_classes
281 comp_classes
= plugin
.filter_component_classes
283 if comp_spec
.class_name
not in comp_classes
:
284 cc_type
= 'source' if comp_cls_type
== _CompClsType
.SOURCE
else 'filter'
286 'no such {} component class in "{}" plugin: {}'.format(
287 cc_type
, comp_spec
.plugin_name
, comp_spec
.class_name
291 comp_cls
= comp_classes
[comp_spec
.class_name
]
292 name
= self
._get
_unique
_comp
_name
(comp_spec
)
293 comp
= self
._graph
.add_component(
294 comp_cls
, name
, comp_spec
.params
, comp_spec
.obj
, comp_spec
.logging_level
298 def _get_free_muxer_input_port(self
):
299 for port
in self
._muxer
_comp
.input_ports
.values():
300 if not port
.is_connected
:
303 def _connect_src_comp_port(self
, component
, port
):
304 # if this trace collection iterator is in stream intersection
305 # mode, we need this connection:
307 # port -> trimmer -> muxer
312 if self
._stream
_intersection
_mode
:
313 trimmer_comp
= self
._create
_stream
_intersection
_trimmer
(component
, port
)
314 self
._graph
.connect_ports(port
, trimmer_comp
.input_ports
['in'])
315 port_to_muxer
= trimmer_comp
.output_ports
['out']
319 self
._graph
.connect_ports(port_to_muxer
, self
._get
_free
_muxer
_input
_port
())
321 def _graph_port_added(self
, component
, port
):
322 if not self
._connect
_ports
:
325 if type(port
) is bt2_port
._InputPort
:
328 if component
not in [comp
.comp
for comp
in self
._src
_comps
_and
_specs
]:
329 # do not care about non-source components (muxer, trimmer, etc.)
332 self
._connect
_src
_comp
_port
(component
, port
)
334 def _build_graph(self
):
335 self
._graph
= bt2
.Graph()
336 self
._graph
.add_port_added_listener(self
._graph
_port
_added
)
337 self
._muxer
_comp
= self
._create
_muxer
()
339 if self
._begin
_ns
is not None or self
._end
_ns
is not None:
340 trimmer_comp
= self
._create
_trimmer
(self
._begin
_ns
, self
._end
_ns
, 'trimmer')
341 self
._graph
.connect_ports(
342 self
._muxer
_comp
.output_ports
['out'], trimmer_comp
.input_ports
['in']
344 last_flt_out_port
= trimmer_comp
.output_ports
['out']
346 last_flt_out_port
= self
._muxer
_comp
.output_ports
['out']
348 # create extra filter components (chained)
349 for comp_spec
in self
._flt
_comp
_specs
:
350 comp
= self
._create
_comp
(comp_spec
, _CompClsType
.FILTER
)
351 self
._flt
_comps
_and
_specs
.append(_ComponentAndSpec(comp
, comp_spec
))
353 # connect the extra filter chain
354 for comp_and_spec
in self
._flt
_comps
_and
_specs
:
355 in_port
= list(comp_and_spec
.comp
.input_ports
.values())[0]
356 out_port
= list(comp_and_spec
.comp
.output_ports
.values())[0]
357 self
._graph
.connect_ports(last_flt_out_port
, in_port
)
358 last_flt_out_port
= out_port
360 # Here we create the components, self._graph_port_added() is
361 # called when they add ports, but the callback returns early
362 # because self._connect_ports is False. This is because the
363 # self._graph_port_added() could not find the associated source
364 # component specification in self._src_comps_and_specs because
365 # it does not exist yet (it needs the created component to
367 for comp_spec
in self
._src
_comp
_specs
:
368 comp
= self
._create
_comp
(comp_spec
, _CompClsType
.SOURCE
)
369 self
._src
_comps
_and
_specs
.append(_ComponentAndSpec(comp
, comp_spec
))
371 # Now we connect the ports which exist at this point. We allow
372 # self._graph_port_added() to automatically connect _new_ ports.
373 self
._connect
_ports
= True
375 for comp_and_spec
in self
._src
_comps
_and
_specs
:
376 # Keep a separate list because comp_and_spec.output_ports
377 # could change during the connection of one of its ports.
378 # Any new port is handled by self._graph_port_added().
379 out_ports
= [port
for port
in comp_and_spec
.comp
.output_ports
.values()]
381 for out_port
in out_ports
:
382 if out_port
.is_connected
:
385 self
._connect
_src
_comp
_port
(comp_and_spec
.comp
, out_port
)
387 # Add the proxy sink, passing our message list to share consumed
388 # messages with this trace collection message iterator.
389 sink
= self
._graph
.add_component(
390 _TraceCollectionMessageIteratorProxySink
, 'proxy-sink', obj
=self
._msg
_list
392 sink_in_port
= sink
.input_ports
['in']
394 # connect last filter to proxy sink
395 self
._graph
.connect_ports(last_flt_out_port
, sink_in_port
)