Aggregate

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

@@ -28,6 +28,8 @@
  */
 package org.janelia.saalfeldlab.n5.readdata;
 
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Comparator;
 
 /**
@@ -61,7 +63,7 @@ default long end() {
 	}
 
 	static boolean equals(final Range r0, final Range r1) {
-		if (r0 == null && r1==null) {
+		if (r0 == null && r1 == null) {
 			return true;
 		} else if (r0 == null || r1 == null) {
 			return false;
@@ -119,4 +121,45 @@ public String toString() {
 
 		return new DefaultRange(offset, length);
 	}
+
+	/**
+	 * Returns a potentially new collection of {@link Range}s such that 
+	 * adjacent or overlapping Ranges are combined. 
+	 * <p>
+	 * If the input ranges are non-adjacent the input instance is returned, but if aggregation 
+	 * occurs, the result will be a new, sorted List.
+	 *
+	 * @param ranges
+	 * 		a collection of Ranges
+	 *
+	 * @return collection with adjacent or overlapping ranges merged
+	 */
+	static Collection<? extends Range> aggregate(final Collection<? extends Range> ranges) {
+
+		if (ranges.size() == 0)
+			return ranges;
+
+		final ArrayList<Range> sortedRanges = new ArrayList<>(ranges);
+		sortedRanges.sort(Range.COMPARATOR);
+
+		final ArrayList<Range> result = new ArrayList<>();
+		Range lo = null;
+		for (Range hi : sortedRanges) {
+
+			if (lo == null)
+				lo = hi;
+			else if (lo.end() >= hi.offset()) {
+				// merge
+				lo = Range.at(lo.offset(), Math.max(lo.end(), hi.end()) - lo.offset());
+			} else {
+				result.add(lo);
+				lo = hi;
+			}
+		}
+		result.add(lo);
+
+		final boolean wereMerges = ranges.size() != result.size();
+		return wereMerges ? result : ranges;
+	}
+
 }

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.AggregatingPrefetchLazyRead;
 
 /**
  * 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 AggregatingPrefetchLazyRead(lazyRead);
+		return new LazyReadData(aggregatingLazyRead);
 	}
 
 }

src/main/java/org/janelia/saalfeldlab/n5/readdata/prefetch/AggregatingPrefetchLazyRead.java

@@ -0,0 +1,44 @@
+package org.janelia.saalfeldlab.n5.readdata.prefetch;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+import java.util.List;
+import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
+import org.janelia.saalfeldlab.n5.readdata.LazyRead;
+import org.janelia.saalfeldlab.n5.readdata.Range;
+
+/**
+ * A {@link SliceTrackingLazyRead} that implements {@link #prefetch} to
+ * aggregate overlapping / adjacent ranges and then materialize each aggregated
+ * range.
+ */
+public class AggregatingPrefetchLazyRead extends SliceTrackingLazyRead {
+
+	public AggregatingPrefetchLazyRead(final LazyRead delegate) {
+		super(delegate);
+	}
+
+	/**
+	 * Indicates that the given slices will be subsequently read.
+	 * <p>
+	 * This implementation groups overlapping / adjacent {@link Range}s into single read requests.
+	 *
+	 * @param ranges
+	 * 		slice ranges to prefetch
+	 *
+	 * @throws N5IOException
+	 * 		if any I/O error occurs
+	 */
+	@Override
+	public void prefetch(final Collection<? extends Range> ranges) throws N5IOException {
+
+		final List<Range> filteredRanges = new ArrayList<>(ranges);
+		filteredRanges.removeIf(this::isCovered);
+		final Collection<? extends Range> aggregatedRanges = Range.aggregate(filteredRanges);
+		for (final Range slice : aggregatedRanges) {
+			materialize(slice.offset(), slice.length());
+		}
+	}
+
+}

src/main/java/org/janelia/saalfeldlab/n5/readdata/prefetch/EnclosingPrefetchLazyRead.java

@@ -0,0 +1,49 @@
+package org.janelia.saalfeldlab.n5.readdata.prefetch;
+
+import java.util.Collection;
+import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
+import org.janelia.saalfeldlab.n5.readdata.LazyRead;
+import org.janelia.saalfeldlab.n5.readdata.Range;
+
+/**
+ * A {@link SliceTrackingLazyRead} that implements {@link #prefetch} to
+ * materialize the bounding range of all requested ranges.
+ */
+public class EnclosingPrefetchLazyRead extends SliceTrackingLazyRead {
+
+	public EnclosingPrefetchLazyRead(final LazyRead delegate) {
+		super(delegate);
+	}
+
+	/**
+	 * Indicates that the given slices will be subsequently read.
+	 * {@code LazyRead} implementations (optionally) may take steps to prepare
+	 * for these subsequent slices.
+	 * <p>
+	 * Minimal implementation: Find offset and length covering all ranges that
+	 * are not yet fully covered by existing slices. Then materialize the slice
+	 * covering that range.
+	 *
+	 * @param ranges
+	 * 		slice ranges to prefetch
+	 *
+	 * @throws N5IOException
+	 * 		if any I/O error occurs
+	 */
+	@Override
+	public void prefetch(final Collection<? extends Range> ranges) throws N5IOException {
+
+		long fromIndex = Long.MAX_VALUE;
+		long toIndex = Long.MIN_VALUE;
+		for (final Range slice : ranges) {
+			if (!isCovered(slice)) {
+				fromIndex = Math.min(fromIndex, slice.offset());
+				toIndex = Math.max(toIndex, slice.end());
+			}
+		}
+
+		if (fromIndex < toIndex) {
+			materialize(fromIndex, toIndex - fromIndex);
+		}
+	}
+}

src/main/java/org/janelia/saalfeldlab/n5/readdata/prefetch/SliceTrackingLazyRead.java

@@ -0,0 +1,90 @@
+package org.janelia.saalfeldlab.n5.readdata.prefetch;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
+import org.janelia.saalfeldlab.n5.readdata.LazyRead;
+import org.janelia.saalfeldlab.n5.readdata.Range;
+import org.janelia.saalfeldlab.n5.readdata.ReadData;
+
+/**
+ * A {@link LazyRead} that wraps a delegate {@code LazyRead} and keeps track of
+ * all slices that have been {@link #materialize materialized}.
+ * <p>
+ * When materializing a new slice, we first check whether it is completely
+ * covered by a materialized slice that we already track. If so, then we just
+ * return a slice on the existing materialized slice. If not, we materialize the
+ * slice from the delegate track it.
+ */
+public class SliceTrackingLazyRead implements LazyRead {
+
+	protected static class Slice implements Range {
+
+		// Offset and length in the delegate
+		private final long offset;
+		private final long length;
+
+		// Data of this slice
+		private final ReadData data;
+
+		Slice(final long offset, final long length, final ReadData data) {
+			this.offset = offset;
+			this.length = length;
+			this.data = data;
+		}
+
+		@Override
+		public long offset() {
+			return offset;
+		}
+
+		@Override
+		public long length() {
+			return length;
+		}
+
+		@Override
+		public String toString() {
+			return "{" + offset + ", " + length + '}';
+		}
+	}
+
+	protected final List<Slice> slices = new ArrayList<>();
+
+	/**
+	 * The {@code LazyRead} providing our data.
+	 */
+	private final LazyRead delegate;
+
+	public SliceTrackingLazyRead(final LazyRead delegate) {
+		this.delegate = delegate;
+	}
+
+	@Override
+	public void close() throws IOException {
+		delegate.close();
+	}
+
+	@Override
+	public ReadData materialize(final long offset, final long length) throws N5IOException {
+		final Slice containing = Slices.findContainingSlice(slices, offset, length);
+		if (containing != null) {
+			return containing.data.slice(offset - containing.offset, length);
+		} else {
+			final ReadData data = delegate.materialize(offset, length);
+			Slices.addSlice(slices, new Slice(offset, length, data));
+			return data;
+		}
+	}
+
+	@Override
+	public long size() throws N5IOException {
+		return delegate.size();
+	}
+
+	protected boolean isCovered(final Range slice) {
+
+		return Slices.findContainingSlice(slices, slice) != null;
+	}
+}

src/main/java/org/janelia/saalfeldlab/n5/readdata/prefetch/Slices.java

@@ -0,0 +1,116 @@
+package org.janelia.saalfeldlab.n5.readdata.prefetch;
+
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import org.janelia.saalfeldlab.n5.readdata.Range;
+
+class Slices {
+
+	private Slices() {
+		// utility class. should not be instantiated.
+	}
+
+	/**
+	 * In an ordered list of {@code slices}, find a slice that completely contains the given range.
+	 * <p>
+	 * Pre-conditions:
+	 * <ol>
+	 * <li>Slices are ordered by offset.</li>
+	 * <li>If two slices overlap, no slice is fully contained within the other.
+	 *     (Therefore, if {@code a.offset < b.offset} then {@code a.end < b.end}.)</li>
+	 * </ol>
+	 *
+	 * @param slices
+	 * 		ordered list of slices
+	 * @param offset
+	 * 		start of the range to cover
+	 * @param length
+	 * 		length of the range to cover
+	 *
+	 * @return a slice that completely contains the requested range, or {@code null} if no such slice exists
+	 */
+	static <T extends Range> T findContainingSlice(final List<T> slices, final long offset, final long length) {
+
+		// Find the slice with the largest slice.offset <= offset.
+		final int i = Collections.binarySearch(slices, Range.at(offset, 0), Comparator.comparingLong(Range::offset));
+
+		// Largest index of a slice with slice.offset <= offset.
+		final int index = i < 0 ? -i - 2 : i;
+		if (index < 0) {
+			// We find no overlapping slice, because
+			// slices[0].offset is already too large.
+			return null;
+		}
+
+		final T slice = slices.get(index);
+		if (slice.end() < offset + length) {
+			return null;
+		}
+
+		return slice;
+	}
+
+	/**
+	 * In an ordered list of {@code slices}, find a slice that completely contains the given range.
+	 * <p>
+	 * Pre-conditions:
+	 * <ol>
+	 * <li>Slices are ordered by offset.</li>
+	 * <li>If two slices overlap, no slice is fully contained within the other.
+	 *     (Therefore, if {@code a.offset < b.offset} then {@code a.end < b.end}.)</li>
+	 * </ol>
+	 *
+	 * @param slices
+	 * 		ordered list of slices
+	 * @param range
+	 * 		range to cover
+	 *
+	 * @return a slice that completely contains the requested range, or {@code null} if no such slice exists
+	 */
+	static <T extends Range> T findContainingSlice(final List<T> slices, final Range range) {
+		return findContainingSlice(slices, range.offset(), range.length());
+	}
+
+	/**
+	 * Add a new {@code slice} to the {@code slice} list.
+	 * <p>
+	 * Note, that the new {@code slice} is expected to not be fully contained in
+	 * an existing slice!
+	 * <p>
+	 * Pre/post-conditions:
+	 * <ol>
+	 * <li>Slices are ordered by offset.</li>
+	 * <li>If two slices overlap, no slice is fully contained within the other.
+	 *     (Therefore, if {@code a.offset < b.offset} then {@code a.end < b.end}.)</li>
+	 * </ol>
+	 * <p>
+	 * The new {@code slice} will be inserted into the list at the correct position
+	 * (such that {@code slices} remains ordered by slice offset), and all existing
+	 * slices that are fully contained in the new {@code slice} will be removed.
+	 *
+	 * @param slices
+	 * 		ordered list of slices
+	 * @param slice
+	 * 		slice to be inserted
+	 */
+	static <T extends Range> void addSlice(final List<T> slices, final T slice) {
+
+		final int i = Collections.binarySearch(slices, slice, Comparator.comparingLong(Range::offset));
+		final int from = i < 0 ? -i - 1 : i;
+
+		int to = from;
+		while (to < slices.size() && slices.get(to).end() <= slice.end()) {
+			++to;
+		}
+
+		if (from == to) {
+			// empty range: just insert
+			slices.add(from, slice);
+		} else {
+			// overwrite the first element in range, remove the rest
+			slices.set(from, slice);
+			slices.subList(from + 1, to).clear();
+		}
+	}
+}