Added faster indexed seeks.

5 files changed, 205 insertions, 23 deletions

diff --git a/lttng/org.eclipse.linuxtools.ctf.core.tests/src/org/eclipse/linuxtools/ctf/core/tests/headless/ReadTrace.java b/lttng/org.eclipse.linuxtools.ctf.core.tests/src/org/eclipse/linuxtools/ctf/core/tests/headless/ReadTrace.java
index bc58479..d3977e2 100644
--- a/lttng/org.eclipse.linuxtools.ctf.core.tests/src/org/eclipse/linuxtools/ctf/core/tests/headless/ReadTrace.java
+++ b/lttng/org.eclipse.linuxtools.ctf.core.tests/src/org/eclipse/linuxtools/ctf/core/tests/headless/ReadTrace.java
@@ -34,35 +34,45 @@ public class ReadTrace {
         // Change this to enable text output
         final boolean USE_TEXT = false;
 
+        final int LOOP_COUNT = 1;
+
         // Work variables
         Long nbEvent = 0L;
         Vector<Double> benchs = new Vector<Double>();
         CTFTrace trace = null;
         long start, stop;
-        for (int loops = 0; loops < 100; loops++) {
+        for (int loops = 0; loops < LOOP_COUNT; loops++) {
             try {
                 nbEvent = 0L;
                 trace = new CTFTrace(TRACE_PATH);
             } catch (CTFReaderException e) {
-                nbEvent = (long) -1;
+                // do nothing
             }
-
+            long prev = -1;
             start = System.nanoTime();
-            if (nbEvent != -1) {
+            if (USE_TEXT) {
+                System.out.println("Event, " + " Time, " + " type, " + " CPU ");
+            }
+            if (trace != null) {
                 CTFTraceReader traceReader = new CTFTraceReader(trace);
 
                 start = System.nanoTime();
+
                 while (traceReader.hasMoreEvents()) {
                     EventDefinition ed = traceReader.getCurrentEventDef();
                     nbEvent++;
+                    if (prev == traceReader.getIndex()) {
+                        System.out.println("Error on events " + prev);
+                    }
+                    prev = traceReader.getIndex();
                     if (USE_TEXT) {
                         String output = formatDate(ed.timestamp
                                 + trace.getOffset());
-                        System.out.println("Event " + nbEvent + " Time "
-                                + output + " type "
-                                + ed.getDeclaration().getName() + " on CPU "
-                                + ed.getCPU());
+                        System.out.println(traceReader.getIndex() + ", "
+                                + output + ", " + ed.getDeclaration().getName()
+                                + ", " + ed.getCPU());
                     }
+
                     traceReader.advance();
                 }
             }
@@ -78,8 +88,9 @@ public class ReadTrace {
         }
         avg /= benchs.size();
         System.out.println("Time to read = " + avg + " events/ns");
-        for(Double val:benchs){
-            System.out.print(val );System.out.print( ", ");
+        for (Double val : benchs) {
+            System.out.print(val);
+            System.out.print(", ");
         }
 
     }
diff --git a/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/CTFTraceReader.java b/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/CTFTraceReader.java
index df1a5d1..deabd94 100644
--- a/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/CTFTraceReader.java
+++ b/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/CTFTraceReader.java
@@ -19,6 +19,7 @@ import java.util.Vector;
 
 import org.eclipse.linuxtools.ctf.core.event.EventDefinition;
 import org.eclipse.linuxtools.internal.ctf.core.Activator;
+ 
 /**
  * Reads the events of a trace.
  */
@@ -59,6 +60,13 @@ public class CTFTraceReader {
      */
     private long endTime;
 
+    /**
+     * Current event index
+     */
+    private long index;
+
+    private long startIndex;
+
     // ------------------------------------------------------------------------
     // Constructors
     // ------------------------------------------------------------------------
@@ -88,6 +96,7 @@ public class CTFTraceReader {
          */
         this.startTime = prio.peek().getCurrentEvent().timestamp;
         this.endTime = this.startTime;
+        this.index = 0;
     }
 
     /**
@@ -115,6 +124,13 @@ public class CTFTraceReader {
         return this.startTime;
     }
 
+    /**
+     * @return the index
+     */
+    public long getIndex() {
+        return index;
+    }
+
     // ------------------------------------------------------------------------
     // Operations
     // ------------------------------------------------------------------------
@@ -193,7 +209,7 @@ public class CTFTraceReader {
      *         of the trace.
      */
     public EventDefinition getCurrentEventDef() {
-        StreamInputReader top = this.prio.peek();
+        StreamInputReader top = getTopStream();
 
         return (top != null) ? top.getCurrentEvent() : null;
     }
@@ -205,6 +221,9 @@ public class CTFTraceReader {
      */
     public boolean advance() {
         /*
+         * Index the
+         */
+        /*
          * Remove the reader from the top of the priority queue.
          */
         StreamInputReader top = this.prio.poll();
@@ -228,7 +247,21 @@ public class CTFTraceReader {
             this.endTime = Math.max(topEnd, this.endTime);
             this.eventCountPerTraceFile[top.getName()]++;
         }
-
+        if(hasMoreEvents())
+        {
+            /*
+             * increment the index
+             */
+            index++;
+            StreamInputPacketReader packetReader = top.getPacketReader();
+            if (packetReader.hasMoreEvents() == false) {
+                StreamInputPacketIndexEntry currentPacket = packetReader
+                        .getCurrentPacket();
+                currentPacket.indexBegin = startIndex;
+                currentPacket.rankEnd = index;
+                startIndex = index + 1;
+            }
+        }
         /*
          * If there is no reader in the queue, it means the trace reader reached
          * the end of the trace.
@@ -272,12 +305,36 @@ public class CTFTraceReader {
          * Remove all the trace readers from the priority queue
          */
         this.prio.clear();
+        index = 0;
+        for (StreamInputReader streamInputReader : this.streamInputReaders) {
+            /*
+             * Seek the trace reader.
+             */
+            long offset = streamInputReader.seek(timestamp);
+
+            /*
+             * Add it to the priority queue if there is a current event.
+             */
+            if (streamInputReader.getCurrentEvent() != null) {
+                this.prio.add(streamInputReader);
+                index = Math.max(index, streamInputReader.getPacketReader()
+                        .getCurrentPacket().indexBegin + offset);
+            }
+        }
 
+        return hasMoreEvents();
+    }
+
+    public boolean seekIndex( long index )
+    {
+        this.prio.clear();
+        this.index = index;
+        long tempIndex = Long.MAX_VALUE;
         for (StreamInputReader streamInputReader : this.streamInputReaders) {
             /*
              * Seek the trace reader.
              */
-            streamInputReader.seek(timestamp);
+            tempIndex = Math.min(tempIndex, streamInputReader.seekBeforeIndex(index));
 
             /*
              * Add it to the priority queue if there is a current event.
@@ -286,10 +343,20 @@ public class CTFTraceReader {
                 this.prio.add(streamInputReader);
             }
         }
+        /*
+         *  advance for offset
+         */
+        for( long pos = tempIndex; (pos < index) && hasMoreEvents(); pos++){
+            this.advance();
+        }
 
         return hasMoreEvents();
     }
 
+    public StreamInputReader getTopStream() {
+        return this.prio.peek();
+    }
+
     /**
      * Does the trace have more events?
      *
@@ -328,8 +395,8 @@ public class CTFTraceReader {
             int len = (width * this.eventCountPerTraceFile[se.getName()])
                     / numEvents;
 
-            StringBuilder sb = new StringBuilder(
-                    se.getStreamInput().getFilename() + "\t["); //$NON-NLS-1$
+            StringBuilder sb = new StringBuilder(se.getStreamInput()
+                    .getFilename() + "\t["); //$NON-NLS-1$
 
             for (int i = 0; i < len; i++) {
                 sb.append('+');
@@ -355,8 +422,8 @@ public class CTFTraceReader {
         result = (prime * result) + (int) (endTime ^ (endTime >>> 32));
         result = (prime * result) + (int) (startTime ^ (startTime >>> 32));
         result = (prime * result)
-                + ((streamInputReaders == null) ? 0
-                        : streamInputReaders.hashCode());
+                + ((streamInputReaders == null) ? 0 : streamInputReaders
+                        .hashCode());
         result = (prime * result) + ((trace == null) ? 0 : trace.hashCode());
         return result;
     }
diff --git a/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputPacketIndex.java b/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputPacketIndex.java
index eae9dc4..ed044ea 100644
--- a/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputPacketIndex.java
+++ b/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputPacketIndex.java
@@ -12,7 +12,6 @@
 
 package org.eclipse.linuxtools.ctf.core.trace;
 
-import java.util.Collection;
 import java.util.ListIterator;
 import java.util.Vector;
 
@@ -37,7 +36,7 @@ public class StreamInputPacketIndex {
     // Getters/Setters/Predicates
     // ------------------------------------------------------------------------
 
-    public Collection<StreamInputPacketIndexEntry> getEntries() {
+    public Vector<StreamInputPacketIndexEntry> getEntries() {
         return this.entries;
     }
 
@@ -139,5 +138,63 @@ public class StreamInputPacketIndex {
 
         return this.entries.listIterator(guessI);
     }
+    /**
+     * Given a rank, this methods returns the first PacketIndexEntry that
+     * could include the rank, that is the last packet with a begin
+     * rank smaller than the given rank.
+     *
+     * @param index
+     *            The rank to look for.
+     * @return The StreamInputPacketEntry that corresponds to the packet that
+     *         includes the given timestamp.
+     */
+    public ListIterator<StreamInputPacketIndexEntry> searchIndex(final long index) {
+        /*
+         * Start with min and max covering all the elements.
+         */
+        int max = this.entries.size() - 1;
+        int min = 0;
+
+        int guessI;
+        StreamInputPacketIndexEntry guessEntry = null;
+
+        if (index < 0) {
+            throw new IllegalArgumentException("rank is negative"); //$NON-NLS-1$
+        }
+
+        for (;;) {
+            /*
+             * Guess in the middle of min and max. The +1 is so that in case
+             * (min + 1 == max), we choose the packet at the subscript "max"
+             * instead of the one at "min". Otherwise, it would give an infinite
+             * loop.
+             */
+            guessI = (max + min + 1) / 2;
+            guessEntry = this.entries.get(guessI);
+
+            /*
+             * If we reached the point where we focus on a single packet, our
+             * search is done.
+             */
+            if (min == max) {
+                break;
+            }
+
+            if (index < guessEntry.indexBegin) {
+                /*
+                 * If the timestamp if before the begin timestamp, we know that
+                 * the packet to return is before the guess.
+                 */
+                max = guessI - 1;
+            } else if (index >= guessEntry.indexBegin) {
+                /*
+                 * If the timestamp is after the begin timestamp, we know that
+                 * the packet to return is after the guess or is the guess.
+                 */
+                min = guessI;
+            }
+        }
 
+        return this.entries.listIterator(guessI);
+    }
 }
diff --git a/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputPacketIndexEntry.java b/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputPacketIndexEntry.java
index 167d8ff..4efd3ee 100644
--- a/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputPacketIndexEntry.java
+++ b/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputPacketIndexEntry.java
@@ -53,6 +53,11 @@ public class StreamInputPacketIndexEntry {
      */
     public long timestampEnd = 0;
 
+
+    public long indexBegin = 0;
+
+    public long rankEnd = 0;
+
     // ------------------------------------------------------------------------
     // Constructors
     // ------------------------------------------------------------------------
@@ -84,6 +89,10 @@ public class StreamInputPacketIndexEntry {
         return (ts >= timestampBegin) && (ts <= timestampEnd);
     }
 
+    boolean includesRank(long rank){
+        return (rank >= indexBegin) && (rank <= rankEnd);
+    }
+
     @Override
     public String toString() {
         /* Only for debugging, shouldn't be externalized */
diff --git a/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputReader.java b/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputReader.java
index 25a3b2b..a2592d3 100644
--- a/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputReader.java
+++ b/lttng/org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/StreamInputReader.java
@@ -43,6 +43,7 @@ public class StreamInputReader {
      */
     private ListIterator<StreamInputPacketIndexEntry> packetIndexIt;
 
+
     /**
      * Reference to the current event of this trace file (iow, the last on that
      * was read, the next one to be returned)
@@ -140,8 +141,8 @@ public class StreamInputReader {
      * Change the current packet of the packet reader to the next one.
      */
     private void goToNextPacket() {
-        if (this.packetIndexIt.hasNext()) {
-            this.packetReader.setCurrentPacket(this.packetIndexIt.next());
+        if (getPacketIndexIt().hasNext()) {
+            this.packetReader.setCurrentPacket(getPacketIndexIt().next());
         } else {
             this.packetReader.setCurrentPacket(null);
         }
@@ -154,7 +155,8 @@ public class StreamInputReader {
      * @param timestamp
      *            The timestamp to seek to.
      */
-    public void seek(long timestamp) {
+    public long seek(long timestamp) {
+        long offset = 0 ;
         /*
          * Search in the index for the packet to search in.
          */
@@ -175,14 +177,36 @@ public class StreamInputReader {
         while (!done && (this.getCurrentEvent().timestamp < timestamp)) {
             readNextEvent();
             done = (this.getCurrentEvent() == null);
+            offset++ ;
         }
+        return offset;
+    }
+
+
+    public long seekBeforeIndex(long index)
+    {
+        /*
+         * Search in the index for the packet to search in.
+         */
+        this.packetIndexIt = this.streamInput.getIndex().searchIndex(index);
+        /*
+         * Switch to this packet.
+         */
+        goToNextPacket();
+        /*
+         * get the current index
+         */
+        return this.packetReader.getCurrentPacket().indexBegin;
+
     }
 
     public void goToLastEvent() throws CTFReaderException {
         /*
          * Search in the index for the packet to search in.
          */
-        this.packetIndexIt = this.streamInput.getIndex().search(Long.MAX_VALUE);
+        int len = this.streamInput.getIndex().getEntries().size();
+        StreamInputPacketIndexEntry entry = this.streamInput.getIndex().getEntries().get(len-1);
+        seek(entry.timestampEnd - 1 );
         /*
          * Go until the end of that packet
          */
@@ -195,4 +219,18 @@ public class StreamInputReader {
         this.currentEvent = currentEvent;
     }
 
+    /**
+     * @return the packetIndexIt
+     */
+    private ListIterator<StreamInputPacketIndexEntry> getPacketIndexIt() {
+        return packetIndexIt;
+    }
+
+    /**
+     * @return the packetReader
+     */
+    public StreamInputPacketReader getPacketReader() {
+        return packetReader;
+    }
+
 }