refactor: migrate previous batching/aggregate work to current development

feat: use the aggregate/prefetch logic with VolatileReadData.

By default, `VolatileReadData.from` now uses AggregatingSliceTrackingLazyRead. This is useful to ensure the default VolatileReadData behaves reasonably even for reading many chunks from the same shard.

Author: cmhulbert

src/main/java/org/janelia/saalfeldlab/n5/readdata/VolatileReadData.java

@@ -1,7 +1,7 @@
 package org.janelia.saalfeldlab.n5.readdata;
 
-import java.io.InputStream;
 import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
+import org.janelia.saalfeldlab.n5.readdata.prefetch.AggregatingSliceTrackingLazyRead;
 
 /**
  * During its life-time, the content of a {@code VolatileReadData} should not be
@@ -29,7 +29,8 @@ public interface VolatileReadData extends ReadData, AutoCloseable {
 	 * @return a new VolatileReadData
 	 */
 	static VolatileReadData from(final LazyRead lazyRead) {
-		return new LazyReadData(lazyRead);
+		final LazyRead aggregatingLazyRead = new AggregatingSliceTrackingLazyRead(lazyRead);
+		return new LazyReadData(aggregatingLazyRead);
 	}
 
 }

src/test/java/org/janelia/saalfeldlab/n5/backward/CompatibilityTest.java

@@ -35,7 +35,6 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.URI;
 import java.nio.file.Files;
 import java.util.Arrays;

src/test/java/org/janelia/saalfeldlab/n5/benchmarks/ReadDataBenchmarks.java

@@ -29,8 +29,6 @@
 package org.janelia.saalfeldlab.n5.benchmarks;
 
 import java.io.IOException;
-import java.io.OutputStream;
-import java.nio.file.FileSystems;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
@@ -40,7 +38,6 @@
 
 import org.janelia.saalfeldlab.n5.FileSystemKeyValueAccess;
 import org.janelia.saalfeldlab.n5.KeyValueAccess;
-import org.janelia.saalfeldlab.n5.LockedChannel;
 import org.janelia.saalfeldlab.n5.N5Exception;
 import org.janelia.saalfeldlab.n5.readdata.ReadData;
 import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;

src/test/java/org/janelia/saalfeldlab/n5/http/HttpKeyValueAccessTest.java

@@ -30,17 +30,14 @@
 
 import org.apache.commons.io.IOUtils;
 import org.janelia.saalfeldlab.n5.HttpKeyValueAccess;
-import org.janelia.saalfeldlab.n5.LockedChannel;
 import org.janelia.saalfeldlab.n5.N5Exception;
 import org.janelia.saalfeldlab.n5.readdata.ReadData;
 import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;
 import org.junit.Test;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.URI;
 import java.nio.charset.Charset;
-import java.util.function.Function;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThrows;

src/test/java/org/janelia/saalfeldlab/n5/kva/TrackingKeyValueAccess.java

@@ -2,16 +2,18 @@
 
 import org.janelia.saalfeldlab.n5.KeyValueAccess;
 import org.janelia.saalfeldlab.n5.N5Exception;
+import org.janelia.saalfeldlab.n5.readdata.DelegatingVolatileReadData;
 import org.janelia.saalfeldlab.n5.readdata.LazyRead;
 import org.janelia.saalfeldlab.n5.readdata.ReadData;
 import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;
-import org.janelia.saalfeldlab.n5.shard.ShardTest;
+import org.janelia.saalfeldlab.n5.readdata.prefetch.AggregatingSliceTrackingLazyRead;
 
 public class TrackingKeyValueAccess extends DelegateKeyValueAccess {
 
     public int numMaterializeCalls = 0;
     public int numIsFileCalls = 0;
     public long totalBytesRead = 0;
+    public boolean aggregate = false;
 
     public TrackingKeyValueAccess(final KeyValueAccess kva) {
         super(kva);
@@ -25,15 +27,27 @@ public boolean isFile(String normalPath) {
 
     @Override
     public VolatileReadData createReadData(final String normalPath) {
-//			throw new N5NoSuchKeyException("Test No Such Key");
-        return VolatileReadData.from(new TrackingVolatileReadData(kva.createReadData(normalPath)));
+
+        final VolatileReadData volatileReadData = kva.createReadData(normalPath);
+        final TrackingLazyRead trackingLazyRead = new TrackingLazyRead(volatileReadData);
+        LazyRead lazyRead = trackingLazyRead;
+        if (aggregate)
+            lazyRead = new AggregatingSliceTrackingLazyRead(trackingLazyRead);
+        VolatileReadData delegate = VolatileReadData.from( lazyRead );
+        return new DelegatingVolatileReadData(delegate) {
+            @Override
+            public void close() throws N5Exception.N5IOException {
+                super.close(); //closes delegate
+                volatileReadData.close();
+            }
+        };
     }
 
-    private class TrackingVolatileReadData implements LazyRead {
+    private class TrackingLazyRead implements LazyRead {
 
         private final VolatileReadData readData;
 
-        TrackingVolatileReadData(final VolatileReadData readData) {
+        TrackingLazyRead(final VolatileReadData readData) {
             this.readData = readData;
         }
 

src/test/java/org/janelia/saalfeldlab/n5/readdata/DelegatingVolatileReadData.java

@@ -0,0 +1,82 @@
+package org.janelia.saalfeldlab.n5.readdata;
+
+import org.janelia.saalfeldlab.n5.N5Exception;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+
+public class DelegatingVolatileReadData implements VolatileReadData {
+
+    private final VolatileReadData delegate;
+
+    public DelegatingVolatileReadData(VolatileReadData delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public void close() throws N5Exception.N5IOException {
+        delegate.close();
+    }
+
+    @Override
+    public long length() {
+        return delegate.length();
+    }
+
+    @Override
+    public long requireLength() throws N5Exception.N5IOException {
+        return delegate.requireLength();
+    }
+
+    @Override
+    public ReadData limit(long length) throws N5Exception.N5IOException {
+        return delegate.limit(length);
+    }
+
+    @Override
+    public ReadData slice(long offset, long length) throws N5Exception.N5IOException {
+        return delegate.slice(offset, length);
+    }
+
+    @Override
+    public ReadData slice(Range range) throws N5Exception.N5IOException {
+        return delegate.slice(range);
+    }
+
+    @Override
+    public InputStream inputStream() throws N5Exception.N5IOException, IllegalStateException {
+        return delegate.inputStream();
+    }
+
+    @Override
+    public byte[] allBytes() throws N5Exception.N5IOException, IllegalStateException {
+        return delegate.allBytes();
+    }
+
+    @Override
+    public ByteBuffer toByteBuffer() throws N5Exception.N5IOException, IllegalStateException {
+        return delegate.toByteBuffer();
+    }
+
+    @Override
+    public ReadData materialize() throws N5Exception.N5IOException {
+        return delegate.materialize();
+    }
+
+    @Override
+    public void writeTo(OutputStream outputStream) throws N5Exception.N5IOException, IllegalStateException {
+        delegate.writeTo(outputStream);
+    }
+
+    @Override
+    public void prefetch(Collection<? extends Range> ranges) throws N5Exception.N5IOException {
+        delegate.prefetch(ranges);
+    }
+
+    @Override
+    public ReadData encode(OutputStreamOperator encoder) {
+        return delegate.encode(encoder);
+    }
+}

src/test/java/org/janelia/saalfeldlab/n5/readdata/ReadDataTests.java

@@ -38,7 +38,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.nio.file.FileSystems;
 import java.util.Arrays;
 import java.util.function.IntUnaryOperator;
 

src/test/java/org/janelia/saalfeldlab/n5/shard/ShardTest.java

@@ -537,20 +537,17 @@ public void numReadsTest() {
 				new ByteArrayDataBlock(chunkSize, new long[]{11, 11}, data)
 		);
 
-		writer.resetNumMaterializeCalls();
-		writer.readChunks(dataset, datasetAttributes, Collections.singletonList(new long[] {0,0}));
-		System.out.println(writer.getNumMaterializeCalls());
+        writer.resetNumMaterializeCalls();
+        writer.readChunks(dataset, datasetAttributes, Collections.singletonList(new long[] {0,0}));
 
 		ArrayList<long[]> ptList = new ArrayList<>();
 		ptList.add(new long[] {0, 0});
 		ptList.add(new long[] {0, 1});
 		ptList.add(new long[] {1, 0});
 		ptList.add(new long[] {1, 1});
 
-		writer.resetNumMaterializeCalls();
-		writer.readChunks(dataset, datasetAttributes, ptList);
-		System.out.println(writer.getNumMaterializeCalls());
-		System.out.println("");
+        writer.resetNumMaterializeCalls();
+        writer.readChunks(dataset, datasetAttributes, ptList);
 	}
 
     @Test
@@ -656,14 +653,29 @@ public void testPartialReadAggregationBehavior() {
 			writer.resetNumMaterializeCalls();
 			writer.readChunks(dataset, datasetAttributes, ptList);
 
-			// TODO change this if and when we implement aggregation of read calls
-			// one for the index, one for each of the four blocks
-			assertEquals(5, writer.getNumMaterializeCalls());
+			// one for the index, one for the four blocks (aggregated)
+			assertEquals(2, writer.getNumMaterializeCalls());
 
 			writer.resetNumMaterializeCalls();
 			writer.readBlock(dataset, datasetAttributes, new long[] {0,0});
-			// one for the index, one for each of the four blocks
-			assertEquals(5, writer.getNumMaterializeCalls());
+			// one for the index, one for the four blocks (aggregated)
+			assertEquals(2, writer.getNumMaterializeCalls());
+
+
+			/**
+			 *  Aggregate read calls
+			 */
+            writer.tkva.aggregate = true;
+			writer.resetNumMaterializeCalls();
+			writer.readChunks(dataset, datasetAttributes, ptList);
+
+			// one for the index, one that covers ALL the blocks)
+			assertEquals(2, writer.getNumMaterializeCalls());
+
+			writer.resetNumMaterializeCalls();
+			writer.readBlock(dataset, datasetAttributes, new long[] {0,0});
+			// one for the index, one that covers ALL the blocks
+			assertEquals(2, writer.getNumMaterializeCalls());
 		}
 	}