Optimize last-position sequence lookup

Author: peter-lawrey

src/main/java/net/openhft/chronicle/queue/impl/single/SCQIndexing.java

@@ -505,7 +505,8 @@ private ScanResult linearScan0(@NotNull final Wire wire,
             fromKnownIndex = lastIndex;
         }
 
-        bytes.readPositionUnlimited(knownAddress);
+        bytes.readLimitToCapacity();
+        bytes.readPosition(knownAddress);
 
         for (long i = fromKnownIndex; ; i++) {
             try {
@@ -565,10 +566,26 @@ long linearScanByPosition(@NotNull final Wire wire,
                               final long indexOfNext,
                               final long startAddress,
                               boolean inclusive) throws EOFException {
+        return linearScanByPosition(wire, toPosition, indexOfNext, startAddress, inclusive, "linearScan by position");
+    }
+
+    /**
+     * Variant that lets the caller supply a description for the perf log so different scan
+     * origins (fast path from {@code writePosition}, fall-through from indexed anchor, etc.)
+     * are distinguishable in logs -- important because the whole point of the
+     * {@code MAX_VALUE} fast path is to <em>avoid</em> the brute-force fall-through scan, and
+     * we want to be able to verify that from log output rather than guess.
+     */
+    long linearScanByPosition(@NotNull final Wire wire,
+                              final long toPosition,
+                              final long indexOfNext,
+                              final long startAddress,
+                              boolean inclusive,
+                              String desc) throws EOFException {
         long start = REPORT_LINEAR_SCAN ? System.nanoTime() : 0;
         long index = linearScanByPosition0(wire, toPosition, indexOfNext, startAddress, inclusive);
         if (REPORT_LINEAR_SCAN) {
-            printLinearScanTime(index, startAddress, start, "linearScan by position");
+            printLinearScanTime(index, startAddress, start, desc);
         }
         return index;
     }
@@ -664,12 +681,13 @@ long linearScanByPosition0(@NotNull final Wire wire,
      * @return The starting index for the scan.
      */
     private long calculateInitialValue(long toPosition, long indexOfNext, long startAddress, Bytes<?> bytes, long lastAddress, long lastIndex) {
+        bytes.readLimit(bytes.capacity());
         if (lastAddress > 0 && toPosition == lastAddress
                 && lastIndex != Sequence.NOT_FOUND && lastIndex != Sequence.NOT_FOUND_RETRY) {
-            bytes.readPositionUnlimited(toPosition);
+            bytes.readPosition(toPosition);
             return lastIndex - 1;
         } else {
-            bytes.readPositionUnlimited(startAddress);
+            bytes.readPosition(startAddress);
             return indexOfNext - 1;
         }
     }
@@ -709,6 +727,44 @@ public long nextEntryToBeIndexed() {
         return nextEntryToBeIndexed.getVolatileValue();
     }
 
+    long sequenceForMaxPosition(@NotNull ExcerptContext ec,
+                                boolean inclusive) throws StreamCorruptedException {
+        int retry = 0;
+        for (; retry < 128; retry++) {
+            long lastWritePos = writePosition.getVolatileValue();
+            long latestSeq = sequence.getSequence(lastWritePos);
+            if (latestSeq >= 0) {
+                if (retry > 0) {
+                    Jvm.perf().on(getClass(),
+                            "sequenceForPosition(MAX_VALUE) tracker-read retried" + " " + retry + " times");
+                }
+                try {
+                    return linearScanByPosition(ec.wireForIndex(), Long.MAX_VALUE, latestSeq, lastWritePos, inclusive,
+                            "linearScan from writePosition (sequenceForPosition fast path)");
+                } catch (EOFException e) {
+                    throw new UncheckedIOException(e);
+                }
+            }
+            if (latestSeq == Sequence.NOT_FOUND)
+                break;
+
+            // NOT_FOUND_RETRY: writer raced. Spin-retry for the first couple of iterations
+            // (the race window is sub-microsecond); after that yield to let the writer make
+            // progress instead of starving them with our volatile reads.
+            if (retry > 2)
+                Thread.yield();
+        }
+        if (retry >= 128) {
+            // Retry budget exhausted in a row of NOT_FOUND_RETRY -- surface this clearly:
+            // it means we're about to do a brute-force indexed-anchor scan instead of the
+            // efficient writePos-anchored fast-path scan. The log makes a regression visible.
+            Jvm.perf().on(getClass(),
+                    "sequenceForPosition(MAX_VALUE) tracker-read retry loop exhausted" + " after " + retry +
+                            " attempts; falling through to indexed lookup");
+        }
+        return sequenceForPosition(ec, Long.MAX_VALUE, inclusive);
+    }
+
     /**
      * Returns the sequence number for a given position in the wire.
      * If an exact match is found for the position, the corresponding index is returned;
@@ -773,7 +829,7 @@ long sequenceForPosition(@NotNull ExcerptContext ec,
         }
         try {
             // Perform a linear scan if no exact match is found.
-            return linearScanByPosition(wire, position, indexOfNext, lastKnownAddress, inclusive);
+            return linearScanByPosition(wire, position, indexOfNext, lastKnownAddress, inclusive, "linearScan from indexed anchor (sequenceForPosition fall-through)");
         } catch (EOFException e) {
             throw new UncheckedIOException(e);
         }
@@ -976,13 +1032,16 @@ public long lastSequenceNumber(@NotNull ExcerptContext ec)
                     break;
                 try {
                     Wire wireForIndex = ec.wireForIndex();
-                    return wireForIndex == null ? sequence : linearScanByPosition(wireForIndex, Long.MAX_VALUE, sequence, address, true);
+                    return wireForIndex == null ? sequence : linearScanByPosition(wireForIndex, Long.MAX_VALUE, sequence, address, true,
+                            "linearScan from writePosition (lastSequenceNumber tail-check)");
                 } catch (EOFException e) {
                     throw new UncheckedIOException(e);
                 }
             }
         }
 
+        // The write-position retry budget has already been spent above. Fall through
+        // directly to the indexed lookup instead of restarting the MAX_VALUE fast path.
         return sequenceForPosition(ec, Long.MAX_VALUE, false);
     }
 

src/main/java/net/openhft/chronicle/queue/impl/single/SingleChronicleQueueStore.java

@@ -377,7 +377,9 @@ public MappedBytes bytes() {
     public long sequenceForPosition(@NotNull final ExcerptContext ec, final long position, boolean inclusive) throws StreamCorruptedException {
         throwExceptionIfClosed();
 
-        return indexing.sequenceForPosition(ec, position, inclusive);
+        return position == Long.MAX_VALUE
+                ? indexing.sequenceForMaxPosition(ec, inclusive)
+                : indexing.sequenceForPosition(ec, position, inclusive);
     }
 
     /**