- * Reads the events of a trace file. + * A CTF trace event reader. Reads the events of a trace file. + * + * @version 1.0 + * @author Matthew Khouzam + * @author Simon Marchi */ -public class StreamInputReader { +public class StreamInputReader implements AutoCloseable { // ------------------------------------------------------------------------ // Attributes @@ -33,31 +34,32 @@ public class StreamInputReader { /** * The StreamInput we are reading. */ - private final StreamInput streamInput; + private final StreamInput fStreamInput; /** * The packet reader used to read packets from this trace file. */ - private final StreamInputPacketReader packetReader; + private final StreamInputPacketReader fPacketReader; /** * Iterator on the packet index */ - private int packetIndex; + private int fPacketIndex; /** * Reference to the current event of this trace file (iow, the last on that * was read, the next one to be returned) */ - private EventDefinition currentEvent = null; - - private int name; + private EventDefinition fCurrentEvent = null; - private CTFTraceReader parent; + private int fId; - private final long prevIndex; + private CTFTraceReader fParent; - private final boolean isthisthefirsttimewerereadingthisgivenstream; + /** + * Live trace reading + */ + private boolean fLive = false; // ------------------------------------------------------------------------ // Constructors @@ -68,60 +70,130 @@ public class StreamInputReader { * * @param streamInput * The StreamInput to read. + * @throws CTFReaderException + * if an error occurs + * @since 2.0 */ - public StreamInputReader(StreamInput streamInput) { - this.streamInput = streamInput; - this.packetReader = new StreamInputPacketReader(this); - this.isthisthefirsttimewerereadingthisgivenstream = true; + public StreamInputReader(StreamInput streamInput) throws CTFReaderException { + fStreamInput = streamInput; + fPacketReader = new StreamInputPacketReader(this); /* * Get the iterator on the packet index. */ - this.packetIndex = 0; - - this.prevIndex = 0; + fPacketIndex = 0; /* * Make first packet the current one. */ goToNextPacket(); } + /** + * Dispose the StreamInputReader + * + * @since 3.0 + */ + @Override + public void close() { + fPacketReader.close(); + } + // ------------------------------------------------------------------------ // Getters/Setters/Predicates // ------------------------------------------------------------------------ + /** + * Gets the current event in this stream + * + * @return the current event in the stream, null if the stream is + * finished/empty/malformed + */ public EventDefinition getCurrentEvent() { - return this.currentEvent; - } - - public StructDefinition getCurrentPacketContext() { - return this.packetReader.getStreamPacketContextDef(); + return fCurrentEvent; } + /** + * Gets the byte order for a trace + * + * @return the trace byte order + */ public ByteOrder getByteOrder() { - return streamInput.getStream().getTrace().getByteOrder(); + return fStreamInput.getStream().getTrace().getByteOrder(); } + /** + * Gets the name of the stream (it's an id and a number) + * + * @return gets the stream name (it's a number) + */ public int getName() { - return this.name; + return fId; } + /** + * Sets the name of the stream + * + * @param name + * the name of the stream, (it's a number) + */ public void setName(int name) { - this.name = name; + fId = name; } + /** + * Gets the CPU of a stream. It's the same as the one in /proc or running + * the asm CPUID instruction + * + * @return The CPU id (a number) + */ public int getCPU() { - return this.packetReader.getCPU(); + return fPacketReader.getCPU(); } + /** + * Gets the filename of the stream being read + * + * @return The filename of the stream being read + */ public String getFilename() { - return streamInput.getFilename(); + return fStreamInput.getFilename(); } /* * for internal use only */ StreamInput getStreamInput() { - return streamInput; + return fStreamInput; + } + + /** + * Set the trace to live mode + * + * @param live + * whether the trace is read live or not + * @since 3.0 + */ + public void setLive(boolean live) { + fLive = live; + } + + /** + * Get if the trace is to read live or not + * + * @return whether the trace is live or not + * @since 3.0 + */ + public boolean isLive() { + return fLive; + } + + /** + * Get the event context of the stream + * + * @return the event context declaration of the stream + * @since 3.0 + */ + public StructDeclaration getStreamEventContextDecl() { + return getStreamInput().getStream().getEventContextDecl(); } // ------------------------------------------------------------------------ @@ -131,63 +203,54 @@ public class StreamInputReader { * Reads the next event in the current event variable. * * @return If an event has been successfully read. + * @throws CTFReaderException + * if an error occurs + * @since 3.0 */ - public boolean readNextEvent() { - + public CTFResponse readNextEvent() throws CTFReaderException { /* * Change packet if needed */ - if (!this.packetReader.hasMoreEvents()) { - final StreamInputPacketIndexEntry prevPacket = this.packetReader + if (!fPacketReader.hasMoreEvents()) { + final StreamInputPacketIndexEntry prevPacket = fPacketReader .getCurrentPacket(); - if (prevPacket != null) { + if (prevPacket != null || fLive) { goToNextPacket(); - final StreamInputPacketIndexEntry currentPacket = this.packetReader - .getCurrentPacket(); } + } /* * If an event is available, read it. */ - if (this.packetReader.hasMoreEvents()) { - try { - this.setCurrentEvent(this.packetReader.readNextEvent()); - } catch (CTFReaderException e) { - /* - * Some problem happened, we'll assume that there are no more - * events - */ - return false; - } - return true; + if (fPacketReader.hasMoreEvents()) { + setCurrentEvent(fPacketReader.readNextEvent()); + return CTFResponse.OK; } - final StreamInputPacketIndexEntry currentPacket = this - .getPacketReader().getCurrentPacket(); this.setCurrentEvent(null); - return false; + return fLive ? CTFResponse.WAIT : CTFResponse.FINISH; } /** * Change the current packet of the packet reader to the next one. + * + * @throws CTFReaderException + * if an error occurs */ - private void goToNextPacket() { - packetIndex++; - if (getPacketSize() >= (packetIndex + 1)) { - this.packetReader.setCurrentPacket(getPacket()); + private void goToNextPacket() throws CTFReaderException { + fPacketIndex++; + // did we already index the packet? + if (getPacketSize() >= (fPacketIndex + 1)) { + fPacketReader.setCurrentPacket(getPacket()); } else { - try { - if (this.streamInput.addPacketHeaderIndex()) { - packetIndex = getPacketSize() - 1; - this.packetReader.setCurrentPacket(getPacket()); - - } else { - this.packetReader.setCurrentPacket(null); - } - - } catch (CTFReaderException e) { - this.packetReader.setCurrentPacket(null); + // go to the next packet if there is one, index it at the same time + if (fStreamInput.addPacketHeaderIndex()) { + fPacketIndex = getPacketSize() - 1; + fPacketReader.setCurrentPacket(getPacket()); + } else { + // out of packets + fPacketReader.setCurrentPacket(null); } } } @@ -196,33 +259,48 @@ public class StreamInputReader { * @return */ private int getPacketSize() { - return streamInput.getIndex().getEntries().size(); + return fStreamInput.getIndex().getEntries().size(); } /** * Changes the location of the trace file reader so that the current event - * is the first event with a timestamp greater than the given timestamp. + * is the first event with a timestamp greater or equal the given timestamp. * * @param timestamp * The timestamp to seek to. + * @return The offset compared to the current position + * @throws CTFReaderException + * if an error occurs */ - public long seek(long timestamp) { + public long seek(long timestamp) throws CTFReaderException { long offset = 0; - /* - * Search in the index for the packet to search in. - */ - this.packetIndex = this.streamInput.getIndex().search(timestamp) - .previousIndex(); + + gotoPacket(timestamp); /* - * Switch to this packet. + * index up to the desired timestamp. */ - goToNextPacket(); + while ((fPacketReader.getCurrentPacket() != null) + && (fPacketReader.getCurrentPacket().getTimestampEnd() < timestamp)) { + try { + fStreamInput.addPacketHeaderIndex(); + goToNextPacket(); + } catch (CTFReaderException e) { + // do nothing here + } + } + if (fPacketReader.getCurrentPacket() == null) { + gotoPacket(timestamp); + } /* - * Advance until A. we reached the end of the trace file (which means - * the given timestamp is after the last event), or B. we found the - * first event with a timestamp greater than the given timestamp. + * Advance until either of these conditions are met: + * + * - reached the end of the trace file (the given timestamp is after the + * last event) + * + * - found the first event with a timestamp greater or equal the given + * timestamp. */ readNextEvent(); boolean done = (this.getCurrentEvent() == null); @@ -234,13 +312,33 @@ public class StreamInputReader { return offset; } - public void goToLastEvent() { + /** + * @param timestamp + * the time to seek + * @throws CTFReaderException + * if an error occurs + */ + private void gotoPacket(long timestamp) throws CTFReaderException { + fPacketIndex = fStreamInput.getIndex().search(timestamp) + .previousIndex(); + /* + * Switch to this packet. + */ + goToNextPacket(); + } + + /** + * Seeks the last event of a stream and returns it. + * + * @throws CTFReaderException + * if an error occurs + */ + public void goToLastEvent() throws CTFReaderException { /* * Search in the index for the packet to search in. */ - final int len = this.streamInput.getIndex().getEntries().size(); + final int len = fStreamInput.getIndex().getEntries().size(); - StreamInputPacketIndexEntry entry = null; /* * Go to beginning of trace. */ @@ -248,7 +346,7 @@ public class StreamInputReader { /* * if the trace is empty. */ - if ((len == 0) || (this.packetReader.hasMoreEvents() == false)) { + if ((len == 0) || (fPacketReader.hasMoreEvents() == false)) { /* * This means the trace is empty. abort. */ @@ -258,9 +356,9 @@ public class StreamInputReader { * Go to the last packet that contains events. */ for (int pos = len - 1; pos > 0; pos--) { - packetIndex = pos; - this.packetReader.setCurrentPacket(getPacket()); - if (this.packetReader.hasMoreEvents()) { + fPacketIndex = pos; + fPacketReader.setCurrentPacket(getPacket()); + if (fPacketReader.hasMoreEvents()) { break; } } @@ -269,8 +367,8 @@ public class StreamInputReader { * Go until the end of that packet */ EventDefinition prevEvent = null; - while (this.currentEvent != null) { - prevEvent = this.currentEvent; + while (fCurrentEvent != null) { + prevEvent = fCurrentEvent; this.readNextEvent(); } /* @@ -283,7 +381,7 @@ public class StreamInputReader { * @return the parent */ public CTFTraceReader getParent() { - return parent; + return fParent; } /** @@ -291,51 +389,47 @@ public class StreamInputReader { * the parent to set */ public void setParent(CTFTraceReader parent) { - this.parent = parent; + fParent = parent; } + /** + * Sets the current event in a stream input reader + * + * @param currentEvent + * the event to set + */ public void setCurrentEvent(EventDefinition currentEvent) { - this.currentEvent = currentEvent; + fCurrentEvent = currentEvent; } /** * @return the packetIndexIt */ private int getPacketIndex() { - return packetIndex; + return fPacketIndex; } private StreamInputPacketIndexEntry getPacket() { - return streamInput.getIndex().getEntries().get(getPacketIndex()); + return fStreamInput.getIndex().getEntries().get(getPacketIndex()); } /** * @return the packetReader */ public StreamInputPacketReader getPacketReader() { - return packetReader; + return fPacketReader; } - /* - * (non-Javadoc) - * - * @see java.lang.Object#hashCode() - */ @Override public int hashCode() { final int prime = 31; int result = 1; - result = (prime * result) + name; + result = (prime * result) + fId; result = (prime * result) - + ((streamInput == null) ? 0 : streamInput.hashCode()); + + ((fStreamInput == null) ? 0 : fStreamInput.hashCode()); return result; } - /* - * (non-Javadoc) - * - * @see java.lang.Object#equals(java.lang.Object) - */ @Override public boolean equals(Object obj) { if (this == obj) { @@ -348,17 +442,22 @@ public class StreamInputReader { return false; } StreamInputReader other = (StreamInputReader) obj; - if (name != other.name) { + if (fId != other.fId) { return false; } - if (streamInput == null) { - if (other.streamInput != null) { + if (fStreamInput == null) { + if (other.fStreamInput != null) { return false; } - } else if (!streamInput.equals(other.streamInput)) { + } else if (!fStreamInput.equals(other.fStreamInput)) { return false; } return true; } + @Override + public String toString() { + // this helps debugging + return fId + ' ' + fCurrentEvent.toString(); + } }