1# Copyright 2013 The Chromium Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4''' TraceEventImporter imports TraceEvent-formatted data
5into the provided model.
6This is a port of the trace event importer from
7https://code.google.com/p/trace-viewer/
8'''
9
10import copy
11import json
12import re
13
14from telemetry.core.timeline import importer
15import telemetry.core.timeline.async_slice as tracing_async_slice
16
17class TraceEventTimelineImporter(importer.TimelineImporter):
18  def __init__(self, model, event_data):
19    super(TraceEventTimelineImporter, self).__init__(
20        model, event_data, import_priority=1)
21
22    self._events_were_from_string = False
23    self._all_async_events = []
24    self._all_object_events = []
25
26    if type(event_data) is str:
27      # If the event data begins with a [, then we know it should end with a ].
28      # The reason we check for this is because some tracing implementations
29      # cannot guarantee that a ']' gets written to the trace file. So, we are
30      # forgiving and if this is obviously the case, we fix it up before
31      # throwing the string at JSON.parse.
32      if event_data[0] == '[':
33        event_data = re.sub(r'[\r|\n]*$', '', event_data)
34        event_data = re.sub(r'\s*,\s*$', '', event_data)
35        if event_data[-1] != ']':
36          event_data = event_data + ']'
37
38      self._events = json.loads(event_data)
39      self._events_were_from_string = True
40    else:
41      self._events = event_data
42
43    # Some trace_event implementations put the actual trace events
44    # inside a container. E.g { ... , traceEvents: [ ] }
45    # If we see that, just pull out the trace events.
46    if 'traceEvents' in self._events:
47      container = self._events
48      self._events = self._events['traceEvents']
49      for field_name in container:
50        if field_name == 'traceEvents':
51          continue
52
53        # Any other fields in the container should be treated as metadata.
54        self._model.metadata.append({
55            'name' : field_name,
56            'value' : container['field_name']})
57
58  @staticmethod
59  def CanImport(event_data):
60    ''' Returns whether obj is a TraceEvent array. '''
61    # May be encoded JSON. But we dont want to parse it fully yet.
62    # Use a simple heuristic:
63    #   - event_data that starts with [ are probably trace_event
64    #   - event_data that starts with { are probably trace_event
65    # May be encoded JSON. Treat files that start with { as importable by us.
66    if isinstance(event_data, str):
67      return len(event_data) > 0 and (event_data[0] == '{'
68          or event_data[0] == '[')
69
70    # Might just be an array of events
71    if (isinstance(event_data, list) and len(event_data)
72        and 'ph' in event_data[0]):
73      return True
74
75    # Might be an object with a traceEvents field in it.
76    if 'traceEvents' in event_data:
77      trace_events = event_data.get('traceEvents', None)
78      return (type(trace_events) is list and
79          len(trace_events) > 0 and 'ph' in trace_events[0])
80
81    return False
82
83  def _GetOrCreateProcess(self, pid):
84    return self._model.GetOrCreateProcess(pid)
85
86  def _DeepCopyIfNeeded(self, obj):
87    if self._events_were_from_string:
88      return obj
89    return copy.deepcopy(obj)
90
91  def _ProcessAsyncEvent(self, event):
92    '''Helper to process an 'async finish' event, which will close an
93    open slice.
94    '''
95    thread = (self._GetOrCreateProcess(event['pid'])
96        .GetOrCreateThread(event['tid']))
97    self._all_async_events.append({
98        'event': event,
99        'thread': thread})
100
101  def _ProcessCounterEvent(self, event):
102    '''Helper that creates and adds samples to a Counter object based on
103    'C' phase events.
104    '''
105    if 'id' in event:
106      ctr_name = event['name'] + '[' + str(event['id']) + ']'
107    else:
108      ctr_name = event['name']
109
110    ctr = (self._GetOrCreateProcess(event['pid'])
111        .GetOrCreateCounter(event['cat'], ctr_name))
112    # Initialize the counter's series fields if needed.
113    if len(ctr.series_names) == 0:
114      #TODO: implement counter object
115      for series_name in event['args']:
116        ctr.series_names.append(series_name)
117      if len(ctr.series_names) == 0:
118        self._model.import_errors.append('Expected counter ' + event['name'] +
119            ' to have at least one argument to use as a value.')
120        # Drop the counter.
121        del ctr.parent.counters[ctr.full_name]
122        return
123
124    # Add the sample values.
125    ctr.timestamps.append(event['ts'] / 1000.0)
126    for series_name in ctr.series_names:
127      if series_name not in event['args']:
128        ctr.samples.append(0)
129        continue
130      ctr.samples.append(event['args'][series_name])
131
132  def _ProcessObjectEvent(self, event):
133    thread = (self._GetOrCreateProcess(event['pid'])
134      .GetOrCreateThread(event['tid']))
135    self._all_object_events.append({
136        'event': event,
137        'thread': thread})
138
139  def _ProcessDurationEvent(self, event):
140    thread = (self._GetOrCreateProcess(event['pid'])
141      .GetOrCreateThread(event['tid']))
142    if not thread.IsTimestampValidForBeginOrEnd(event['ts'] / 1000.0):
143      self._model.import_errors.append(
144          'Timestamps are moving backward.')
145      return
146
147    if event['ph'] == 'B':
148      thread.BeginSlice(event['cat'],
149                        event['name'],
150                        event['ts'] / 1000.0,
151                        event['args'])
152    elif event['ph'] == 'E':
153      thread = (self._GetOrCreateProcess(event['pid'])
154        .GetOrCreateThread(event['tid']))
155      if not thread.IsTimestampValidForBeginOrEnd(event['ts'] / 1000.0):
156        self._model.import_errors.append(
157            'Timestamps are moving backward.')
158        return
159      if not thread.open_slice_count:
160        self._model.import_errors.append(
161            'E phase event without a matching B phase event.')
162        return
163
164      new_slice = thread.EndSlice(event['ts'] / 1000.0)
165      for arg_name, arg_value in event.get('args', {}).iteritems():
166        if arg_name in new_slice.args:
167          self._model.import_errors.append(
168              'Both the B and E phases of ' + new_slice.name +
169              ' provided values for argument ' + arg_name + '. ' +
170              'The value of the E phase event will be used.')
171        new_slice.args[arg_name] = arg_value
172
173  def _ProcessMetadataEvent(self, event):
174    if event['name'] == 'thread_name':
175      thread = (self._GetOrCreateProcess(event['pid'])
176          .GetOrCreateThread(event['tid']))
177      thread.name = event['args']['name']
178    if event['name'] == 'process_name':
179      process = self._GetOrCreateProcess(event['pid'])
180      process.name = event['args']['name']
181    else:
182      self._model.import_errors.append(
183          'Unrecognized metadata name: ' + event['name'])
184
185  def _ProcessInstantEvent(self, event):
186    # Treat an Instant event as a duration 0 slice.
187    # SliceTrack's redraw() knows how to handle this.
188    thread = (self._GetOrCreateProcess(event['pid'])
189      .GetOrCreateThread(event['tid']))
190    thread.BeginSlice(event['cat'],
191                      event['name'],
192                      event['ts'] / 1000.0,
193                      event.get('args'))
194    thread.EndSlice(event['ts'] / 1000.0)
195
196  def _ProcessSampleEvent(self, event):
197    thread = (self._GetOrCreateProcess(event['pid'])
198        .GetOrCreateThread(event['tid']))
199    thread.AddSample(event['cat'],
200                     event['name'],
201                     event['ts'] / 1000.0,
202                     args=event.get('args'))
203
204  def ImportEvents(self):
205    ''' Walks through the events_ list and outputs the structures discovered to
206    model_.
207    '''
208    for event in self._events:
209      phase = event.get('ph', None)
210      if phase == 'B' or phase == 'E':
211        self._ProcessDurationEvent(event)
212      elif phase == 'S' or phase == 'F' or phase == 'T':
213        self._ProcessAsyncEvent(event)
214      # Note, I is historic. The instant event marker got changed, but we
215      # want to support loading load trace files so we have both I and i.
216      elif phase == 'I' or phase == 'i':
217        self._ProcessInstantEvent(event)
218      elif phase == 'P':
219        self._ProcessSampleEvent(event)
220      elif phase == 'C':
221        self._ProcessCounterEvent(event)
222      elif phase == 'M':
223        self._ProcessMetadataEvent(event)
224      elif phase == 'N' or phase == 'D' or phase == 'O':
225        self._ProcessObjectEvent(event)
226      elif phase == 's' or phase == 't' or phase == 'f':
227        # NB: toss until there's proper support
228        pass
229      else:
230        self._model.import_errors.append('Unrecognized event phase: ' +
231            phase + '(' + event['name'] + ')')
232
233    return self._model
234
235  def FinalizeImport(self):
236    '''Called by the Model after all other importers have imported their
237    events.'''
238    self._model.UpdateBounds()
239
240    # We need to reupdate the bounds in case the minimum start time changes
241    self._model.UpdateBounds()
242    self._CreateAsyncSlices()
243    self._CreateExplicitObjects()
244    self._CreateImplicitObjects()
245
246  def _CreateAsyncSlices(self):
247    if len(self._all_async_events) == 0:
248      return
249
250    self._all_async_events.sort(
251        cmp=lambda x, y: x['event']['ts'] - y['event']['ts'])
252
253    async_event_states_by_name_then_id = {}
254
255    all_async_events = self._all_async_events
256    for async_event_state in all_async_events:
257      event = async_event_state['event']
258      name = event.get('name', None)
259      if name is None:
260        self._model.import_errors.append(
261            'Async events (ph: S, T or F) require an name parameter.')
262        continue
263
264      event_id = event.get('id')
265      if event_id is None:
266        self._model.import_errors.append(
267            'Async events (ph: S, T or F) require an id parameter.')
268        continue
269
270      # TODO(simonjam): Add a synchronous tick on the appropriate thread.
271
272      if event['ph'] == 'S':
273        if not name in async_event_states_by_name_then_id:
274          async_event_states_by_name_then_id[name] = {}
275        if event_id in async_event_states_by_name_then_id[name]:
276          self._model.import_errors.append(
277              'At ' + event['ts'] + ', a slice of the same id ' + event_id +
278              ' was alrady open.')
279          continue
280
281        async_event_states_by_name_then_id[name][event_id] = []
282        async_event_states_by_name_then_id[name][event_id].append(
283            async_event_state)
284      else:
285        if name not in async_event_states_by_name_then_id:
286          self._model.import_errors.append(
287              'At ' + str(event['ts']) + ', no slice named ' + name +
288              ' was open.')
289          continue
290        if event_id not in async_event_states_by_name_then_id[name]:
291          self._model.import_errors.append(
292              'At ' + str(event['ts']) + ', no slice named ' + name +
293              ' with id=' + event_id + ' was open.')
294          continue
295        events = async_event_states_by_name_then_id[name][event_id]
296        events.append(async_event_state)
297
298        if event['ph'] == 'F':
299          # Create a slice from start to end.
300          async_slice = tracing_async_slice.AsyncSlice(
301              events[0]['event']['cat'],
302              name,
303              events[0]['event']['ts'] / 1000.0)
304
305          async_slice.duration = ((event['ts'] / 1000.0)
306              - (events[0]['event']['ts'] / 1000.0))
307
308          async_slice.start_thread = events[0]['thread']
309          async_slice.end_thread = async_event_state['thread']
310          async_slice.id = event_id
311          async_slice.args = events[0]['event']['args']
312
313          # Create sub_slices for each step.
314          for j in xrange(1, len(events)):
315            sub_name = name
316            if events[j - 1]['event']['ph'] == 'T':
317              sub_name = name + ':' + events[j - 1]['event']['args']['step']
318            sub_slice = tracing_async_slice.AsyncSlice(
319                events[0]['event']['cat'],
320                sub_name,
321                events[j - 1]['event']['ts'] / 1000.0)
322            sub_slice.parent_slice = async_slice
323
324            sub_slice.duration = ((events[j]['event']['ts'] / 1000.0)
325                - (events[j - 1]['event']['ts'] / 1000.0))
326
327            sub_slice.start_thread = events[j - 1]['thread']
328            sub_slice.end_thread = events[j]['thread']
329            sub_slice.id = event_id
330            sub_slice.args = events[j - 1]['event']['args']
331
332            async_slice.AddSubSlice(sub_slice)
333
334          # The args for the finish event go in the last sub_slice.
335          last_slice = async_slice.sub_slices[-1]
336          for arg_name, arg_value in event['args'].iteritems():
337            last_slice.args[arg_name] = arg_value
338
339          # Add |async_slice| to the start-thread's async_slices.
340          async_slice.start_thread.AddAsyncSlice(async_slice)
341          del async_event_states_by_name_then_id[name][event_id]
342
343  def _CreateExplicitObjects(self):
344    # TODO(tengs): Implement object instance parsing
345    pass
346
347  def _CreateImplicitObjects(self):
348    # TODO(tengs): Implement object instance parsing
349    pass
350