1# Copyright 2014 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.
4import telemetry.timeline.async_slice as async_slice_module
5import telemetry.timeline.event_container as event_container
6import telemetry.timeline.flow_event as flow_event_module
7import telemetry.timeline.sample as sample_module
8import telemetry.timeline.slice as slice_module
9
10
11class Thread(event_container.TimelineEventContainer):
12  ''' A Thread stores all the trace events collected for a particular
13  thread. We organize the synchronous slices on a thread by "subrows," where
14  subrow 0 has all the root slices, subrow 1 those nested 1 deep, and so on.
15  The asynchronous slices are stored in an AsyncSliceGroup object.
16  '''
17  def __init__(self, process, tid):
18    super(Thread, self).__init__('thread %s' % tid, parent=process)
19    self.tid = tid
20    self._async_slices = []
21    self._flow_events = []
22    self._samples = []
23    self._toplevel_slices = []
24    self._all_slices = []
25
26    # State only valid during import.
27    self._open_slices = []
28    self._newly_added_slices = []
29
30  @property
31  def toplevel_slices(self):
32    return self._toplevel_slices
33
34  @property
35  def all_slices(self):
36    return self._all_slices
37
38  @property
39  def samples(self):
40    return self._samples
41
42  @property
43  def async_slices(self):
44    return self._async_slices
45
46  @property
47  def open_slice_count(self):
48    return len(self._open_slices)
49
50  def IterChildContainers(self):
51    return
52    yield # pylint: disable=W0101
53
54  def IterEventsInThisContainer(self, event_type_predicate, event_predicate):
55    if event_type_predicate(slice_module.Slice):
56      for s in self._newly_added_slices:
57        if event_predicate(s):
58          yield s
59      for s in self._all_slices:
60        if event_predicate(s):
61          yield s
62
63    if event_type_predicate(async_slice_module.AsyncSlice):
64      for async_slice in self._async_slices:
65        if event_predicate(async_slice):
66          yield async_slice
67        for sub_slice in async_slice.IterEventsInThisContainerRecrusively():
68          if event_predicate(sub_slice):
69            yield sub_slice
70
71    if event_type_predicate(flow_event_module.FlowEvent):
72      for flow_event in self._flow_events:
73        if event_predicate(flow_event):
74          yield flow_event
75
76    if event_type_predicate(sample_module.Sample):
77      for sample in self._samples:
78        if event_predicate(sample):
79          yield sample
80
81  def AddSample(self, category, name, timestamp, args=None):
82    if len(self._samples) and timestamp < self._samples[-1].start:
83      raise ValueError(
84          'Samples must be added in increasing timestamp order')
85    sample = sample_module.Sample(self,
86        category, name, timestamp, args=args)
87    self._samples.append(sample)
88
89  def AddAsyncSlice(self, async_slice):
90    self._async_slices.append(async_slice)
91
92  def AddFlowEvent(self, flow_event):
93    self._flow_events.append(flow_event)
94
95  def BeginSlice(self, category, name, timestamp, thread_timestamp=None,
96                 args=None):
97    """Opens a new slice for the thread.
98    Calls to beginSlice and endSlice must be made with
99    non-monotonically-decreasing timestamps.
100
101    * category: Category to which the slice belongs.
102    * name: Name of the slice to add.
103    * timestamp: The timetsamp of the slice, in milliseconds.
104    * thread_timestamp: Thread specific clock (scheduled) timestamp of the
105                        slice, in milliseconds.
106    * args: Arguments associated with
107
108    Returns newly opened slice
109    """
110    if len(self._open_slices) > 0 and timestamp < self._open_slices[-1].start:
111      raise ValueError(
112          'Slices must be added in increasing timestamp order')
113    new_slice = slice_module.Slice(self, category, name, timestamp,
114                                    thread_timestamp=thread_timestamp,
115                                    args=args)
116    self._open_slices.append(new_slice)
117    new_slice.did_not_finish = True
118    self.PushSlice(new_slice)
119    return new_slice
120
121  def EndSlice(self, end_timestamp, end_thread_timestamp=None):
122    """ Ends the last begun slice in this group and pushes it onto the slice
123    array.
124
125    * end_timestamp: Timestamp when the slice ended in milliseconds
126    * end_thread_timestamp: Timestamp when the scheduled time of the slice ended
127                            in milliseconds
128
129    returns completed slice.
130    """
131    if not len(self._open_slices):
132      raise ValueError(
133          'EndSlice called without an open slice')
134    curr_slice = self._open_slices.pop()
135    if end_timestamp < curr_slice.start:
136      raise ValueError(
137          'Slice %s end time is before its start.' % curr_slice.name)
138    curr_slice.duration = end_timestamp - curr_slice.start
139    if end_thread_timestamp != None:
140      if curr_slice.thread_start == None:
141        raise ValueError(
142            'EndSlice with thread_timestamp called on open slice without ' +
143            'thread_timestamp')
144      curr_slice.thread_duration = (end_thread_timestamp -
145                                    curr_slice.thread_start)
146    curr_slice.did_not_finish = False
147    return curr_slice
148
149  def PushCompleteSlice(self, category, name, timestamp, duration,
150                        thread_timestamp, thread_duration, args=None):
151    new_slice = slice_module.Slice(self, category, name, timestamp,
152                                   thread_timestamp=thread_timestamp,
153                                   args=args)
154    if duration == None:
155      new_slice.did_not_finish = True
156    else:
157      new_slice.duration = duration
158      new_slice.thread_duration = thread_duration
159    self.PushSlice(new_slice)
160    return new_slice
161
162  def PushSlice(self, new_slice):
163    self._newly_added_slices.append(new_slice)
164    return new_slice
165
166  def AutoCloseOpenSlices(self, max_timestamp, max_thread_timestamp):
167    for s in self._newly_added_slices:
168      if s.did_not_finish:
169        s.duration = max_timestamp - s.start
170        assert s.duration >= 0
171        if s.thread_start != None:
172          s.thread_duration = max_thread_timestamp - s.thread_start
173          assert s.thread_duration >= 0
174    self._open_slices = []
175
176  def IsTimestampValidForBeginOrEnd(self, timestamp):
177    if not len(self._open_slices):
178      return True
179    return timestamp >= self._open_slices[-1].start
180
181  def FinalizeImport(self):
182    self._BuildSliceSubRows()
183
184  def _BuildSliceSubRows(self):
185    '''This function works by walking through slices by start time.
186
187     The basic idea here is to insert each slice as deep into the subrow
188     list as it can go such that every subslice is fully contained by its
189     parent slice.
190
191     Visually, if we start with this:
192      0:  [    a       ]
193      1:    [  b  ]
194      2:    [c][d]
195
196     To place this slice:
197                   [e]
198     We first check row 2's last item, [d]. [e] wont fit into [d] (they dont
199     even intersect). So we go to row 1. That gives us [b], and [d] wont fit
200     into that either. So, we go to row 0 and its last slice, [a]. That can
201     completely contain [e], so that means we should add [e] as a subslice
202     of [a]. That puts it on row 1, yielding:
203      0:  [    a       ]
204      1:    [  b  ][e]
205      2:    [c][d]
206
207     If we then get this slice:
208                          [f]
209     We do the same deepest-to-shallowest walk of the subrows trying to fit
210     it. This time, it doesn't fit in any open slice. So, we simply append
211     it to row 0 (a root slice):
212      0:  [    a       ]  [f]
213      1:    [  b  ][e]
214    '''
215    def CompareSlices(s1, s2):
216      if s1.start == s2.start:
217        # Break ties by having the slice with the greatest
218        # end timestamp come first.
219        return cmp(s2.end, s1.end)
220      return cmp(s1.start, s2.start)
221
222    assert len(self._toplevel_slices) == 0
223    assert len(self._all_slices) == 0
224    if not len(self._newly_added_slices):
225      return
226
227    self._all_slices.extend(self._newly_added_slices)
228
229    sorted_slices = sorted(self._newly_added_slices, cmp=CompareSlices)
230    root_slice = sorted_slices[0]
231    self._toplevel_slices.append(root_slice)
232    for s in sorted_slices[1:]:
233      if not self._AddSliceIfBounds(root_slice, s):
234        root_slice = s
235        self._toplevel_slices.append(root_slice)
236    self._newly_added_slices = []
237
238
239  def _AddSliceIfBounds(self, root, child):
240    ''' Adds a child slice to a root slice its proper row.
241    Return False if the child slice is not in the bounds
242    of the root slice.
243
244    Because we know that the start time of child is >= the start time
245    of all other slices seen so far, we can just check the last slice
246    of each row for bounding.
247    '''
248    # The source trace data is in microseconds but we store it as milliseconds
249    # in floating-point. Since we can't represent micros as millis perfectly,
250    # two end=start+duration combos that should be the same will be slightly
251    # different. Round back to micros to ensure equality below.
252    child_end_micros = round(child.end * 1000)
253    root_end_micros =  round(root.end * 1000)
254    if child.start >= root.start and child_end_micros <= root_end_micros:
255      if len(root.sub_slices) > 0:
256        if self._AddSliceIfBounds(root.sub_slices[-1], child):
257          return True
258      child.parent_slice = root
259      root.AddSubSlice(child)
260      return True
261    return False
262