Fix seek of empty stream
[deliverable/tracecompass.git] / org.eclipse.linuxtools.ctf.core / src / org / eclipse / linuxtools / ctf / core / trace / CTFTraceReader.java
1 /*******************************************************************************
2 * Copyright (c) 2011-2012 Ericsson, Ecole Polytechnique de Montreal and others
3 *
4 * All rights reserved. This program and the accompanying materials are made
5 * available under the terms of the Eclipse Public License v1.0 which
6 * accompanies this distribution, and is available at
7 * http://www.eclipse.org/legal/epl-v10.html
8 *
9 * Contributors: Matthew Khouzam - Initial API and implementation
10 * Contributors: Alexandre Montplaisir - Initial API and implementation
11 *******************************************************************************/
12
13 package org.eclipse.linuxtools.ctf.core.trace;
14
15 import java.util.Collection;
16 import java.util.HashMap;
17 import java.util.PriorityQueue;
18 import java.util.Set;
19 import java.util.Vector;
20
21 import org.eclipse.linuxtools.ctf.core.event.EventDefinition;
22 import org.eclipse.linuxtools.internal.ctf.core.Activator;
23 import org.eclipse.linuxtools.internal.ctf.core.trace.Stream;
24 import org.eclipse.linuxtools.internal.ctf.core.trace.StreamInput;
25 import org.eclipse.linuxtools.internal.ctf.core.trace.StreamInputPacketIndexEntry;
26 import org.eclipse.linuxtools.internal.ctf.core.trace.StreamInputReaderTimestampComparator;
27
28 /**
29 * Reads the events of a trace.
30 */
31
32 public class CTFTraceReader {
33
34 // ------------------------------------------------------------------------
35 // Attributes
36 // ------------------------------------------------------------------------
37
38 /**
39 * The trace to read from.
40 */
41 private final CTFTrace trace;
42
43 /**
44 * Vector of all the trace file readers.
45 */
46 private final Vector<StreamInputReader> streamInputReaders = new Vector<StreamInputReader>();
47
48 /**
49 * Priority queue to order the trace file readers by timestamp.
50 */
51 protected PriorityQueue<StreamInputReader> prio;
52
53 /**
54 * Array to count the number of event per trace file.
55 */
56 private int[] eventCountPerTraceFile;
57
58 /**
59 * Timestamp of the first event in the trace
60 */
61 private long startTime;
62
63 /**
64 * Timestamp of the last event read so far
65 */
66 private long endTime;
67
68 /**
69 * Current event index
70 */
71 private long fIndex;
72
73 private final HashMap<Integer, Long> startIndex;
74
75 // ------------------------------------------------------------------------
76 // Constructors
77 // ------------------------------------------------------------------------
78
79 /**
80 * Constructs a TraceReader to read a trace.
81 *
82 * @param trace
83 * The trace to read from.
84 * @throws CTFReaderException
85 */
86 public CTFTraceReader(CTFTrace trace) {
87 this.trace = trace;
88
89 /**
90 * Create the trace file readers.
91 */
92 createStreamInputReaders();
93
94 /**
95 * Populate the timestamp-based priority queue.
96 */
97 populateStreamInputReaderHeap();
98
99 /**
100 * Get the start Time of this trace
101 * bear in mind that the trace could be empty.
102 */
103 this.startTime = 0;// prio.peek().getPacketReader().getCurrentPacket().getTimestampBegin();
104 if (hasMoreEvents()) {
105 this.startTime = prio.peek().getCurrentEvent().getTimestamp();
106 this.endTime = this.startTime;
107 this.fIndex = 0;
108 }
109 startIndex = new HashMap<Integer, Long>();
110 }
111
112 /**
113 * Copy constructor
114 */
115 public CTFTraceReader copyFrom() {
116 CTFTraceReader newReader = null;
117
118 newReader = new CTFTraceReader(this.trace);
119 newReader.startTime = this.startTime;
120 newReader.endTime = this.endTime;
121 return newReader;
122 }
123
124 // ------------------------------------------------------------------------
125 // Getters/Setters/Predicates
126 // ------------------------------------------------------------------------
127
128 /**
129 * Return the start time of this trace (== timestamp of the first event)
130 *
131 * @return the trace start time
132 */
133 public long getStartTime() {
134 return this.startTime;
135 }
136
137 /**
138 * @return the index
139 */
140 public long getIndex() {
141 return fIndex;
142 }
143
144 // ------------------------------------------------------------------------
145 // Operations
146 // ------------------------------------------------------------------------
147
148 /**
149 * Creates one trace file reader per trace file contained in the trace.
150 */
151 private void createStreamInputReaders() {
152 Collection<Stream> streams = this.trace.getStreams().values();
153
154 /*
155 * For each stream.
156 */
157 for (Stream stream : streams) {
158 Set<StreamInput> streamInputs = stream.getStreamInputs();
159
160 /*
161 * For each trace file of the stream.
162 */
163 for (StreamInput streamInput : streamInputs) {
164 /*
165 * Create a reader.
166 */
167 StreamInputReader streamInputReader = new StreamInputReader(
168 streamInput);
169
170 /*
171 * Add it to the group.
172 */
173 this.streamInputReaders.add(streamInputReader);
174 }
175 }
176
177 /*
178 * Create the array to count the number of event per trace file.
179 */
180 this.eventCountPerTraceFile = new int[this.streamInputReaders.size()];
181 }
182
183 /**
184 * Initializes the priority queue used to choose the trace file with the
185 * lower next event timestamp.
186 */
187 private void populateStreamInputReaderHeap() {
188 /*
189 * Create the priority queue with a size twice as bigger as the number
190 * of reader in order to avoid constant resizing.
191 */
192 this.prio = new PriorityQueue<StreamInputReader>(
193 this.streamInputReaders.size() * 2,
194 new StreamInputReaderTimestampComparator());
195
196 int pos = 0;
197
198 for (StreamInputReader reader : this.streamInputReaders) {
199 /*
200 * Add each trace file reader in the priority queue, if we are able
201 * to read an event from it.
202 */
203 if (reader.readNextEvent()) {
204 this.prio.add(reader);
205
206 this.eventCountPerTraceFile[pos] = 0;
207 reader.setName(pos);
208
209 pos++;
210 }
211 }
212 }
213
214 /**
215 * Get the current event, which is the current event of the trace file
216 * reader with the lowest timestamp.
217 *
218 * @return An event definition, or null of the trace reader reached the end
219 * of the trace.
220 */
221 public EventDefinition getCurrentEventDef() {
222 StreamInputReader top = getTopStream();
223
224 return (top != null) ? top.getCurrentEvent() : null;
225 }
226
227 /**
228 * Go to the next event.
229 *
230 * @return True if an event was read.
231 */
232 public boolean advance() {
233 /*
234 * Index the
235 */
236 /*
237 * Remove the reader from the top of the priority queue.
238 */
239 StreamInputReader top = this.prio.poll();
240
241 /*
242 * If the queue was empty.
243 */
244 if (top == null) {
245 return false;
246 }
247 /*
248 * index if needed
249 */
250 if (hasMoreEvents()) {
251 StreamInputPacketReader packetReader = top.getPacketReader();
252 boolean packetHasMoreEvents = packetReader.hasMoreEvents();
253 StreamInputPacketIndexEntry currentPacket = packetReader
254 .getCurrentPacket();
255 if (!packetHasMoreEvents) {
256 int n = this.streamInputReaders.indexOf(top);
257 if (!startIndex.containsKey(n)) {
258 startIndex.put(n, 0L);
259 }
260 currentPacket.setIndexBegin(startIndex.get(n));
261 currentPacket.setIndexEnd(fIndex);
262 startIndex.put(n, fIndex + 1);
263 }
264 }
265 /*
266 * Read the next event of this reader.
267 */
268 if (top.readNextEvent()) {
269 /*
270 * Add it back in the queue.
271 */
272 this.prio.add(top);
273 final long topEnd = top.getCurrentEvent().getTimestamp();
274 this.endTime = Math.max(topEnd, this.endTime);
275 this.eventCountPerTraceFile[top.getName()]++;
276 /*
277 * increment the index
278 */
279 fIndex++;
280 }
281 boolean hasMoreEvents = hasMoreEvents();
282
283 /*
284 * If there is no reader in the queue, it means the trace reader reached
285 * the end of the trace.
286 */
287 return hasMoreEvents;
288 }
289
290 /**
291 * Go to the last event in the trace.
292 *
293 * @throws CTFReaderException
294 */
295 public void goToLastEvent() throws CTFReaderException {
296
297 for (StreamInputReader streamInputReader : this.streamInputReaders) {
298 /*
299 * Seek the trace reader.
300 */
301 streamInputReader.goToLastEvent();
302 }
303 int count = prio.size();
304 for (int i = 0; i < (count - 1); i++) {
305 /*
306 * Cull all the streams aside from the last one
307 */
308 prio.poll();
309 }
310 }
311
312 /**
313 * Seeks to a given timestamp It will go to the event just after the
314 * timestamp or the timestamp itself. if a if a trace is 10 20 30 40 and
315 * you're looking for 19, it'll give you 20, it you want 20, you'll get 20,
316 * if you want 21, you'll get 30. You want -inf, you'll get the first
317 * element, you want +inf, you'll get the end of the file with no events.
318 *
319 * @param timestamp
320 * the timestamp to seek to
321 * @return true if the trace has more events following the timestamp
322 */
323 public boolean seek(long timestamp) {
324 /*
325 * Remove all the trace readers from the priority queue
326 */
327 this.prio.clear();
328 fIndex = 0;
329 long offset = 0;
330 for (StreamInputReader streamInputReader : this.streamInputReaders) {
331 /*
332 * Seek the trace reader.
333 */
334 offset += streamInputReader.seek(timestamp);
335
336 /*
337 * Add it to the priority queue if there is a current event.
338 */
339
340 }
341 for (StreamInputReader streamInputReader : this.streamInputReaders) {
342 if (streamInputReader.getCurrentEvent() != null) {
343 this.prio.add(streamInputReader);
344 fIndex = Math.max(fIndex, streamInputReader.getPacketReader()
345 .getCurrentPacket().getIndexBegin()
346 + offset);
347 }
348 }
349 return hasMoreEvents();
350 }
351
352 public boolean seekIndex(long index) {
353 this.prio.clear();
354
355 long tempIndex = Long.MIN_VALUE;
356 long tempTimestamp = Long.MIN_VALUE;
357 try {
358 for (StreamInputReader streamInputReader : this.streamInputReaders) {
359 /*
360 * Seek the trace reader.
361 */
362 final long streamIndex = streamInputReader.seekIndex(index);
363 if (streamInputReader.getCurrentEvent() != null) {
364 tempIndex = Math.max(tempIndex, streamIndex);
365 EventDefinition currentEvent = streamInputReader
366 .getCurrentEvent();
367 /*
368 * Maybe we're at the beginning of a trace.
369 */
370 if (currentEvent == null) {
371 streamInputReader.readNextEvent();
372 currentEvent = streamInputReader.getCurrentEvent();
373 }
374 if (currentEvent != null) {
375 tempTimestamp = Math.max(tempTimestamp,
376 currentEvent.getTimestamp());
377 } else {
378 /*
379 * probably beyond the last event
380 */
381 tempIndex = goToZero();
382 }
383 }
384
385 }
386 } catch (CTFReaderException e) {
387 /*
388 * Important, if it failed, it's because it's not yet indexed, so we
389 * have to manually advance to the right value.
390 */
391 tempIndex = goToZero();
392 }
393 for (StreamInputReader streamInputReader : this.streamInputReaders) {
394 /*
395 * Add it to the priority queue if there is a current event.
396 */
397
398 if (streamInputReader.getCurrentEvent() != null) {
399 this.prio.add(streamInputReader);
400 }
401 }
402 if (tempIndex == Long.MAX_VALUE) {
403 tempIndex = 0;
404 }
405 long pos = tempIndex;
406 if (index > tempIndex) {
407 /*
408 * advance for offset
409 */
410 while ((prio.peek().getCurrentEvent().getTimestamp() < tempTimestamp)
411 && hasMoreEvents()) {
412 this.advance();
413 }
414
415 for (pos = tempIndex; (pos < index) && hasMoreEvents(); pos++) {
416 this.advance();
417 }
418 }
419 this.fIndex = pos;
420 return hasMoreEvents();
421 }
422
423 /**
424 * Go to the first entry of a trace
425 *
426 * @return 0, the first index.
427 */
428 private long goToZero() {
429 long tempIndex;
430 for (StreamInputReader streamInputReader : this.streamInputReaders) {
431 /*
432 * Seek the trace reader.
433 */
434 streamInputReader.seek(0);
435 }
436 tempIndex = 0;
437 return tempIndex;
438 }
439
440 public StreamInputReader getTopStream() {
441 return this.prio.peek();
442 }
443
444 /**
445 * Does the trace have more events?
446 *
447 * @return true if yes.
448 */
449 public boolean hasMoreEvents() {
450 return this.prio.size() > 0;
451 }
452
453 /**
454 * Prints the event count stats.
455 */
456 public void printStats() {
457 printStats(60);
458 }
459
460 /**
461 * Prints the event count stats.
462 *
463 * @param width
464 * Width of the display.
465 */
466 public void printStats(int width) {
467 int numEvents = 0;
468 if (width == 0) {
469 return;
470 }
471
472 for (int i : this.eventCountPerTraceFile) {
473 numEvents += i;
474 }
475
476 for (int j = 0; j < this.eventCountPerTraceFile.length; j++) {
477 StreamInputReader se = this.streamInputReaders.get(j);
478
479 int len = (width * this.eventCountPerTraceFile[se.getName()])
480 / numEvents;
481
482 StringBuilder sb = new StringBuilder(se.getFilename() + "\t["); //$NON-NLS-1$
483
484 for (int i = 0; i < len; i++) {
485 sb.append('+');
486 }
487
488 for (int i = len; i < width; i++) {
489 sb.append(' ');
490 }
491
492 sb.append("]\t" + this.eventCountPerTraceFile[se.getName()] + " Events"); //$NON-NLS-1$//$NON-NLS-2$
493 Activator.getDefault().log(sb.toString());
494 }
495 }
496
497 public long getEndTime() {
498 return this.endTime;
499 }
500
501 @Override
502 public int hashCode() {
503 final int prime = 31;
504 int result = 1;
505 result = (prime * result) + (int) (endTime ^ (endTime >>> 32));
506 result = (prime * result) + (int) (startTime ^ (startTime >>> 32));
507 result = (prime * result)
508 + ((streamInputReaders == null) ? 0 : streamInputReaders
509 .hashCode());
510 result = (prime * result) + ((trace == null) ? 0 : trace.hashCode());
511 return result;
512 }
513
514 @Override
515 public boolean equals(Object obj) {
516 if (this == obj) {
517 return true;
518 }
519 if (obj == null) {
520 return false;
521 }
522 if (!(obj instanceof CTFTraceReader)) {
523 return false;
524 }
525 CTFTraceReader other = (CTFTraceReader) obj;
526 if (endTime != other.endTime) {
527 return false;
528 }
529 if (startTime != other.startTime) {
530 return false;
531 }
532 if (streamInputReaders == null) {
533 if (other.streamInputReaders != null) {
534 return false;
535 }
536 } else if (!streamInputReaders.equals(other.streamInputReaders)) {
537 return false;
538 }
539 if (trace == null) {
540 if (other.trace != null) {
541 return false;
542 }
543 } else if (!trace.equals(other.trace)) {
544 return false;
545 }
546 return true;
547 }
548
549 /*
550 * (non-Javadoc)
551 *
552 * @see java.lang.Object#toString()
553 */
554 @Override
555 public String toString() {
556 /* Only for debugging, shouldn't be externalized */
557 return "CTFTraceReader [trace=" + trace + ']'; //$NON-NLS-1$
558 }
559
560 public CTFTrace getTrace() {
561 return trace;
562 }
563 }
This page took 0.04375 seconds and 6 git commands to generate.