(uring) Reintroduce auto-slicing of excessively long read batches

code/libraries/array/java/nu/marginalia/array/pool/BufferPool.java

@@ -52,7 +52,7 @@ public class BufferPool implements AutoCloseable {
             throw new RuntimeException(e);
         }
         this.arena = Arena.ofShared();
-        this.pages = new UnsafeMemoryPage[poolSize];
+        this.pages = new MemoryPage[poolSize];
 
         MemorySegment memoryArea = arena.allocate((long) pageSizeBytes*poolSize, 4096);
         for (int i = 0; i < pages.length; i++) {

code/libraries/btree/java/nu/marginalia/skiplist/SkipListReader.java

@@ -90,6 +90,7 @@ public class SkipListReader {
                 }
                 else if (bv == pv) {
                     data.retainAndAdvance();
+                    currentBlockIdx++;
                     continue outer;
                 }
             }
@@ -117,6 +118,8 @@ public class SkipListReader {
 
                 if (!data.hasMore()) {
                     currentBlock += SkipListConstants.BLOCK_SIZE;
+                    currentBlockOffset = 0;
+                    currentBlockIdx = 0;
                 }
                 else {
                     long nextBlock = currentBlock + (long) SkipListConstants.BLOCK_SIZE;
@@ -158,6 +161,8 @@ public class SkipListReader {
 
                     if (!data.hasMore()) {
                         currentBlock += SkipListConstants.BLOCK_SIZE;
+                        currentBlockOffset = 0;
+                        currentBlockIdx = 0;
                     }
                     else {
                         long nextBlock = currentBlock + (long) SkipListConstants.BLOCK_SIZE;
@@ -244,6 +249,8 @@ public class SkipListReader {
 
                     if (pos >= keys.length) {
                         currentBlock += SkipListConstants.BLOCK_SIZE;
+                        currentBlockOffset = 0;
+                        currentBlockIdx = 0;
                     }
                     else {
                         long nextBlock = currentBlock + (long) SkipListConstants.BLOCK_SIZE;
@@ -298,6 +305,7 @@ public class SkipListReader {
                 }
                 else if (bv == pv) {
                     data.rejectAndAdvance();
+                    currentBlockIdx++;
                     continue outer;
                 }
             }
@@ -324,6 +332,8 @@ public class SkipListReader {
                 }
 
                 if (!data.hasMore()) {
+                    currentBlockOffset = 0;
+                    currentBlockIdx = 0;
                     currentBlock += SkipListConstants.BLOCK_SIZE;
                 }
                 else {
@@ -365,6 +375,8 @@ public class SkipListReader {
                         break;
                     }
                     if (!data.hasMore()) {
+                        currentBlockOffset = 0;
+                        currentBlockIdx = 0;
                         currentBlock += SkipListConstants.BLOCK_SIZE;
                     }
                     else {

code/libraries/btree/test/nu/marginalia/skiplist/SkipListReaderTest.java

@@ -102,6 +102,36 @@ public class SkipListReaderTest {
         }
     }
 
+
+    @Test
+    public void testRetainBug() throws IOException {
+        long[] keys = LongStream.range(0, 30000).map(v -> v).toArray();
+        long[] vals = LongStream.range(0, 30000).map(v -> -v).toArray();
+
+        long start;
+        try (var writer = new SkipListWriter(docsFile)) {
+            writer.padDocuments(512);
+            start = writer.writeList(createArray(keys, vals), 0, keys.length);
+        }
+
+        try (var pool = new BufferPool(docsFile, SkipListConstants.BLOCK_SIZE, 8)) {
+            var reader = new SkipListReader(pool, start);
+            long[] values = LongStream.range(0, 4059).toArray();
+            LongQueryBuffer lqb = new LongQueryBuffer(values, values.length);
+
+            reader.retainData(lqb);
+            lqb.finalizeFiltering();
+            System.out.println(Arrays.toString(lqb.copyData()));
+
+            values = LongStream.range(4060, 4070).toArray();
+            lqb = new LongQueryBuffer(values, values.length);
+            reader.retainData(lqb);
+            lqb.finalizeFiltering();
+            System.out.println(Arrays.toString(lqb.copyData()));
+
+        }
+    }
+
     @Test
     public void testRetainFuzz() throws IOException {
 

code/libraries/native/java/nu/marginalia/uring/UringFileReader.java

@@ -66,9 +66,14 @@ public class UringFileReader implements AutoCloseable {
             }
         }
         else {
-            // We *could* break the task into multiple submissions, but this leads to some
-            // very unpredictable read latencies
-            throw new IllegalArgumentException("Submission size exceeds queue size!");
+            for (int i = 0; i < destinations.size(); i+=QUEUE_SIZE) {
+                var destSlice = destinations.subList(i, Math.min(destinations.size(), i+QUEUE_SIZE));
+                var offSlice = offsets.subList(i, Math.min(offsets.size(), i+QUEUE_SIZE));
+                int ret = ring.readBatch(destSlice, offSlice, direct);
+                if (ret != offSlice.size()) {
+                    throw new RuntimeException("Read failed, rv=" + ret);
+                }
+            }
         }
     }