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 import bt2
.message_iterator
28 from collections
import namedtuple
32 # a pair of component and ComponentSpec
33 _ComponentAndSpec
= namedtuple('_ComponentAndSpec', ['comp', 'spec'])
37 def __init__(self
, plugin_name
, component_class_name
, params
=None):
38 utils
._check
_str
(plugin_name
)
39 utils
._check
_str
(component_class_name
)
40 self
._plugin
_name
= plugin_name
41 self
._component
_class
_name
= component_class_name
43 if type(params
) is str:
44 self
._params
= bt2
.create_value({'paths': [params
]})
46 self
._params
= bt2
.create_value(params
)
49 def plugin_name(self
):
50 return self
._plugin
_name
53 def component_class_name(self
):
54 return self
._component
_class
_name
61 # datetime.datetime or integral to nanoseconds
66 if isinstance(obj
, numbers
.Real
):
67 # consider that it's already in seconds
69 elif isinstance(obj
, datetime
.datetime
):
73 raise TypeError('"{}" is not an integral number or a datetime.datetime object'.format(obj
))
83 class TraceCollectionMessageIterator(bt2
.message_iterator
._MessageIterator
):
84 def __init__(self
, source_component_specs
, filter_component_specs
=None,
85 stream_intersection_mode
=False, begin
=None, end
=None):
86 utils
._check
_bool
(stream_intersection_mode
)
87 self
._stream
_intersection
_mode
= stream_intersection_mode
88 self
._begin
_ns
= _get_ns(begin
)
89 self
._end
_ns
= _get_ns(end
)
91 if type(source_component_specs
) is ComponentSpec
:
92 source_component_specs
= [source_component_specs
]
94 if type(filter_component_specs
) is ComponentSpec
:
95 filter_component_specs
= [filter_component_specs
]
96 elif filter_component_specs
is None:
97 filter_component_specs
= []
99 self
._src
_comp
_specs
= source_component_specs
100 self
._flt
_comp
_specs
= filter_component_specs
101 self
._next
_suffix
= 1
102 self
._connect
_ports
= False
104 # lists of _ComponentAndSpec
105 self
._src
_comps
_and
_specs
= []
106 self
._flt
_comps
_and
_specs
= []
108 self
._validate
_component
_specs
(source_component_specs
)
109 self
._validate
_component
_specs
(filter_component_specs
)
112 def _validate_component_specs(self
, comp_specs
):
113 for comp_spec
in comp_specs
:
114 if type(comp_spec
) is not ComponentSpec
:
115 raise TypeError('"{}" object is not a ComponentSpec'.format(type(comp_spec
)))
118 return next(self
._msg
_iter
)
120 def _create_stream_intersection_trimmer(self
, component
, port
):
121 # find the original parameters specified by the user to create
122 # this port's component to get the `path` parameter
123 for src_comp_and_spec
in self
._src
_comps
_and
_specs
:
124 if component
== src_comp_and_spec
.comp
:
128 paths
= src_comp_and_spec
.spec
.params
['paths']
129 except Exception as e
:
130 raise bt2
.Error('all source components must be created with a "paths" parameter in stream intersection mode') from e
132 params
= {'paths': paths
}
134 # query the port's component for the `trace-info` object which
135 # contains the stream intersection range for each exposed
137 query_exec
= bt2
.QueryExecutor()
138 trace_info_res
= query_exec
.query(src_comp_and_spec
.comp
.component_class
,
139 'trace-info', params
)
143 # find the trace info for this port's trace by name's prefix
145 for trace_info
in trace_info_res
:
146 if port
.name
.startswith(str(trace_info
['path'])):
147 range_ns
= trace_info
['intersection-range-ns']
148 begin
= range_ns
['begin']
149 end
= range_ns
['end']
154 if begin
is None or end
is None:
155 raise bt2
.Error('cannot find stream intersection range for port "{}"'.format(port
.name
))
157 name
= 'trimmer-{}-{}'.format(src_comp_and_spec
.comp
.name
, port
.name
)
158 return self
._create
_trimmer
(begin
, end
, name
)
160 def _create_muxer(self
):
161 plugin
= bt2
.find_plugin('utils')
164 raise bt2
.Error('cannot find "utils" plugin (needed for the muxer)')
166 if 'muxer' not in plugin
.filter_component_classes
:
167 raise bt2
.Error('cannot find "muxer" filter component class in "utils" plugin')
169 comp_cls
= plugin
.filter_component_classes
['muxer']
170 return self
._graph
.add_component(comp_cls
, 'muxer')
172 def _create_trimmer(self
, begin_ns
, end_ns
, name
):
173 plugin
= bt2
.find_plugin('utils')
176 raise bt2
.Error('cannot find "utils" plugin (needed for the trimmer)')
178 if 'trimmer' not in plugin
.filter_component_classes
:
179 raise bt2
.Error('cannot find "trimmer" filter component class in "utils" plugin')
183 def ns_to_string(ns
):
184 s_part
= ns
// 1000000000
185 ns_part
= ns
% 1000000000
186 return '{}.{:09d}'.format(s_part
, ns_part
)
188 if begin_ns
is not None:
189 params
['begin'] = ns_to_string(begin_ns
)
191 if end_ns
is not None:
192 params
['end'] = ns_to_string(end_ns
)
194 comp_cls
= plugin
.filter_component_classes
['trimmer']
195 return self
._graph
.add_component(comp_cls
, name
, params
)
197 def _get_unique_comp_name(self
, comp_spec
):
198 name
= '{}-{}'.format(comp_spec
.plugin_name
,
199 comp_spec
.component_class_name
)
200 comps_and_specs
= itertools
.chain(self
._src
_comps
_and
_specs
,
201 self
._flt
_comps
_and
_specs
)
203 if name
in [comp_and_spec
.comp
.name
for comp_and_spec
in comps_and_specs
]:
204 name
+= '-{}'.format(self
._next
_suffix
)
205 self
._next
_suffix
+= 1
209 def _create_comp(self
, comp_spec
, comp_cls_type
):
210 plugin
= bt2
.find_plugin(comp_spec
.plugin_name
)
213 raise bt2
.Error('no such plugin: {}'.format(comp_spec
.plugin_name
))
215 if comp_cls_type
== _CompClsType
.SOURCE
:
216 comp_classes
= plugin
.source_component_classes
218 comp_classes
= plugin
.filter_component_classes
220 if comp_spec
.component_class_name
not in comp_classes
:
221 cc_type
= 'source' if comp_cls_type
== _CompClsType
.SOURCE
else 'filter'
222 raise bt2
.Error('no such {} component class in "{}" plugin: {}'.format(cc_type
,
223 comp_spec
.plugin_name
,
224 comp_spec
.component_class_name
))
226 comp_cls
= comp_classes
[comp_spec
.component_class_name
]
227 name
= self
._get
_unique
_comp
_name
(comp_spec
)
228 comp
= self
._graph
.add_component(comp_cls
, name
, comp_spec
.params
)
231 def _get_free_muxer_input_port(self
):
232 for port
in self
._muxer
_comp
.input_ports
.values():
233 if not port
.is_connected
:
236 def _connect_src_comp_port(self
, component
, port
):
237 # if this trace collection iterator is in stream intersection
238 # mode, we need this connection:
240 # port -> trimmer -> muxer
245 if self
._stream
_intersection
_mode
:
246 trimmer_comp
= self
._create
_stream
_intersection
_trimmer
(component
, port
)
247 self
._graph
.connect_ports(port
, trimmer_comp
.input_ports
['in'])
248 port_to_muxer
= trimmer_comp
.output_ports
['out']
252 self
._graph
.connect_ports(port_to_muxer
, self
._get
_free
_muxer
_input
_port
())
254 def _graph_port_added(self
, component
, port
):
255 if not self
._connect
_ports
:
258 if type(port
) is bt2
.port
._InputPort
:
261 if component
not in [comp
.comp
for comp
in self
._src
_comps
_and
_specs
]:
262 # do not care about non-source components (muxer, trimmer, etc.)
265 self
._connect
_src
_comp
_port
(component
, port
)
267 def _build_graph(self
):
268 self
._graph
= bt2
.Graph()
269 self
._graph
.add_port_added_listener(self
._graph
_port
_added
)
270 self
._muxer
_comp
= self
._create
_muxer
()
272 if self
._begin
_ns
is not None or self
._end
_ns
is not None:
273 trimmer_comp
= self
._create
_trimmer
(self
._begin
_ns
,
274 self
._end
_ns
, 'trimmer')
275 self
._graph
.connect_ports(self
._muxer
_comp
.output_ports
['out'],
276 trimmer_comp
.input_ports
['in'])
277 msg_iter_port
= trimmer_comp
.output_ports
['out']
279 msg_iter_port
= self
._muxer
_comp
.output_ports
['out']
281 # create extra filter components (chained)
282 for comp_spec
in self
._flt
_comp
_specs
:
283 comp
= self
._create
_comp
(comp_spec
, _CompClsType
.FILTER
)
284 self
._flt
_comps
_and
_specs
.append(_ComponentAndSpec(comp
, comp_spec
))
286 # connect the extra filter chain
287 for comp_and_spec
in self
._flt
_comps
_and
_specs
:
288 in_port
= list(comp_and_spec
.comp
.input_ports
.values())[0]
289 out_port
= list(comp_and_spec
.comp
.output_ports
.values())[0]
290 self
._graph
.connect_ports(msg_iter_port
, in_port
)
291 msg_iter_port
= out_port
293 # Here we create the components, self._graph_port_added() is
294 # called when they add ports, but the callback returns early
295 # because self._connect_ports is False. This is because the
296 # self._graph_port_added() could not find the associated source
297 # component specification in self._src_comps_and_specs because
298 # it does not exist yet (it needs the created component to
300 for comp_spec
in self
._src
_comp
_specs
:
301 comp
= self
._create
_comp
(comp_spec
, _CompClsType
.SOURCE
)
302 self
._src
_comps
_and
_specs
.append(_ComponentAndSpec(comp
, comp_spec
))
304 # Now we connect the ports which exist at this point. We allow
305 # self._graph_port_added() to automatically connect _new_ ports.
306 self
._connect
_ports
= True
308 for comp_and_spec
in self
._src
_comps
_and
_specs
:
309 # Keep a separate list because comp_and_spec.output_ports
310 # could change during the connection of one of its ports.
311 # Any new port is handled by self._graph_port_added().
312 out_ports
= [port
for port
in comp_and_spec
.comp
.output_ports
.values()]
314 for out_port
in out_ports
:
315 if out_port
.is_connected
:
318 self
._connect
_src
_comp
_port
(comp_and_spec
.comp
, out_port
)
320 # create this trace collection iterator's message iterator
321 self
._msg
_iter
= self
._graph
.create_output_port_message_iterator(msg_iter_port
)
This page took 0.04086 seconds and 4 git commands to generate.