VolatileReadData

src/main/java/org/janelia/saalfeldlab/n5/FsIoPolicy.java

@@ -0,0 +1,61 @@
+package org.janelia.saalfeldlab.n5;
+
+import org.janelia.saalfeldlab.n5.readdata.ReadData;
+import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;
+
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.file.*;
+
+import static org.janelia.saalfeldlab.n5.FileKeyLockManager.FILE_LOCK_MANAGER;
+
+public class FsIoPolicy {
+
+    static final IoPolicy atomicWithFallback = IoPolicy.withFallback(new Atomic(), new Unsafe());
+
+    public static class Unsafe implements IoPolicy {
+        @Override
+        public void write(String key, ReadData readData) throws IOException {
+            final Path path = Paths.get(key);
+            Files.copy(readData.inputStream(), path, StandardCopyOption.REPLACE_EXISTING);
+        }
+
+        @Override
+        public VolatileReadData read(final String key) throws IOException {
+            final Path path = Paths.get(key);
+            FileSystemKeyValueAccess.FileLazyRead fileLazyRead = new FileSystemKeyValueAccess.FileLazyRead(path, false);
+            return VolatileReadData.from(fileLazyRead);
+        }
+
+        @Override
+        public void delete(final String key) throws IOException {
+            final Path path = Paths.get(key);
+            Files.deleteIfExists(path);
+        }
+    }
+
+    public static class Atomic implements IoPolicy {
+        @Override
+        public void write(String key, ReadData readData) throws IOException {
+            final Path path = Paths.get(key);
+            try (LockedFileChannel channel = FILE_LOCK_MANAGER.lockForWriting(path)) {
+                readData.writeTo(channel.newOutputStream());
+            }
+        }
+
+        @Override
+        public VolatileReadData read(String key) throws IOException {
+            final Path path = Paths.get(key);
+            FileSystemKeyValueAccess.FileLazyRead fileLazyRead = new FileSystemKeyValueAccess.FileLazyRead(path, true);
+            return VolatileReadData.from(fileLazyRead);
+        }
+
+        @Override
+        public void delete(final String key) throws IOException {
+            final Path path = Paths.get(key);
+            try (LockedFileChannel ignore = FILE_LOCK_MANAGER.lockForWriting(path)) {
+                Files.delete(path);
+            }
+        }
+    }
+}

src/main/java/org/janelia/saalfeldlab/n5/GsonKeyValueN5Reader.java

@@ -28,13 +28,12 @@
  */
 package org.janelia.saalfeldlab.n5;
 
-import java.io.IOException;
-import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.UncheckedIOException;
 import java.util.List;
 
 import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
+import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;
 import org.janelia.saalfeldlab.n5.shard.PositionValueAccess;
 
 import com.google.gson.Gson;
@@ -82,11 +81,14 @@ default JsonElement getAttributes(final String pathName) throws N5Exception {
 		final String groupPath = N5URI.normalizeGroupPath(pathName);
 		final String attributesPath = absoluteAttributesPath(groupPath);
 
-		try ( final InputStream in = getKeyValueAccess().createReadData(attributesPath).inputStream() ) {
-			return GsonUtils.readAttributes(new InputStreamReader(in), getGson());
+		try (final VolatileReadData readData = getKeyValueAccess().createReadData(attributesPath);) {
+			if (readData == null) {
+				return null;
+			}
+			return GsonUtils.readAttributes(new InputStreamReader(readData.inputStream()), getGson());
 		} catch (final N5Exception.N5NoSuchKeyException e) {
 			return null;
-		} catch (final IOException | UncheckedIOException | N5IOException e) {
+		} catch (final UncheckedIOException | N5IOException e) {
 			throw new N5IOException("Failed to read attributes from dataset " + pathName, e);
 		}
 	}
@@ -119,6 +121,24 @@ default <T> List<DataBlock<T>> readBlocks(
 		return convertedDatasetAttributes.<T> getDatasetAccess().readBlocks(posKva, blockPositions);
 	}
 
+	@Override
+	default <T> DataBlock<T> readShard(
+			final String pathName,
+			final DatasetAttributes datasetAttributes,
+			final long... gridPosition) throws N5Exception {
+
+		final DatasetAttributes convertedDatasetAttributes = getConvertedDatasetAttributes(datasetAttributes);
+		final int shardLevel = convertedDatasetAttributes.getNestedBlockGrid().numLevels() - 1;
+		try {
+			final PositionValueAccess posKva = PositionValueAccess.fromKva(getKeyValueAccess(), getURI(), N5URI.normalizeGroupPath(pathName),
+					convertedDatasetAttributes);
+			return convertedDatasetAttributes.<T> getDatasetAccess().readShard(posKva, gridPosition, shardLevel);
+
+		} catch (N5Exception.N5NoSuchKeyException e) {
+			return null;
+		}
+	}
+
 	@Override
 	default String[] list(final String pathName) throws N5Exception {
 

src/main/java/org/janelia/saalfeldlab/n5/GsonKeyValueN5Writer.java

@@ -280,7 +280,24 @@ default <T> void writeBlock(
 					"Failed to write block " + Arrays.toString(dataBlock.getGridPosition()) + " into dataset " + path,
 					e);
 		}
+	}
+
+	@Override
+	default <T> void writeShard(
+			final String path,
+			final DatasetAttributes datasetAttributes,
+			final DataBlock<T> shard) throws N5Exception {
 
+		final DatasetAttributes convertedDatasetAttributes = getConvertedDatasetAttributes(datasetAttributes);
+		final int shardLevel = convertedDatasetAttributes.getNestedBlockGrid().numLevels() - 1;
+		try {
+			final PositionValueAccess posKva = PositionValueAccess.fromKva(getKeyValueAccess(), getURI(), N5URI.normalizeGroupPath(path), convertedDatasetAttributes);
+			convertedDatasetAttributes.<T> getDatasetAccess().writeShard(posKva, shard, shardLevel);
+		} catch (final UncheckedIOException e) {
+			throw new N5IOException(
+					"Failed to write block " + Arrays.toString(shard.getGridPosition()) + " into dataset " + path,
+					e);
+		}
 	}
 
 	@Override

src/main/java/org/janelia/saalfeldlab/n5/HttpKeyValueAccess.java

@@ -50,6 +50,8 @@
 import java.nio.channels.NonWritableChannelException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import org.janelia.saalfeldlab.n5.readdata.LazyRead;
+import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;
 
 /**
  * A read-only {@link KeyValueAccess} implementation using HTTP. As a result, calling <code>lockForWriting</code>, <code>createDirectories</code>, or <code>delete</code> will throw an {@link N5Exception}.
@@ -232,8 +234,8 @@ private HttpURLConnection httpRequest(String normalPath, String method) throws I
 	}
 
 	@Override
-	public ReadData createReadData(final String normalPath) {
-		return new KeyValueAccessReadData(new HttpLazyRead(normalPath));
+	public VolatileReadData createReadData(final String normalPath) {
+		return VolatileReadData.from(new HttpLazyRead(normalPath));
 	}
 
 	public LockedChannel lockForReading(final String normalPath) throws N5IOException {
@@ -253,6 +255,12 @@ public LockedChannel lockForWriting(final String normalPath) throws N5IOExceptio
 		throw new N5Exception("HttpKeyValueAccess is read-only");
 	}
 
+	@Override
+	public void write(final String normalPath, final ReadData data) throws N5IOException {
+
+		throw new N5Exception("HttpKeyValueAccess is read-only");
+	}
+
 	/**
 	 * List all 'directory'-like children of a path.
 	 * <p>
@@ -459,6 +467,10 @@ public ReadData materialize(long offset, long length) {
 				throw new N5Exception(e);
 			}
 		}
+
+		@Override
+		public void close() {
+		}
 	}
 
 }

src/main/java/org/janelia/saalfeldlab/n5/IoPolicy.java

@@ -0,0 +1,47 @@
+package org.janelia.saalfeldlab.n5;
+
+import org.janelia.saalfeldlab.n5.readdata.ReadData;
+import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;
+
+import java.io.IOException;
+
+public interface IoPolicy {
+
+    void write(String key, ReadData readData) throws IOException;
+
+    VolatileReadData read(String key) throws IOException;
+
+    void delete(String key) throws IOException;
+
+    static IoPolicy withFallback(IoPolicy primary, IoPolicy fallback) {
+        return new IoPolicy() {
+
+            @Override
+            public void write(String key, ReadData readData) throws IOException {
+                try {
+                    primary.write(key, readData);
+                } catch (IOException e) {
+                    fallback.write(key, readData);
+                }
+            }
+
+            @Override
+            public VolatileReadData read(String key) throws IOException {
+                try {
+                    return primary.read(key);
+                } catch (IOException e) {
+                    return fallback.read(key);
+                }
+            }
+
+            @Override
+            public void delete(String key) throws IOException {
+                try {
+                    primary.delete(key);
+                } catch (IOException e) {
+                    fallback.delete(key);
+                }
+            }
+        };
+    }
+}

src/main/java/org/janelia/saalfeldlab/n5/KeyValueAccess.java

@@ -37,13 +37,13 @@
 import java.util.stream.Collectors;
 
 import org.janelia.saalfeldlab.n5.readdata.ReadData;
+import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;
 
 /**
  * Key value read primitives used by {@link N5KeyValueReader}
  * implementations. This interface implements a subset of access primitives
  * provided by {@link FileSystem} to reduce the implementation burden for
- * backends
- * lacking a {@link FileSystem} implementation (such as AWS-S3).
+ * backends lacking a {@link FileSystem} implementation (such as AWS-S3).
  *
  * @author Stephan Saalfeld
  */
@@ -240,18 +240,44 @@ default URI uri(final String uriString) throws URISyntaxException {
 	boolean isFile( String normalPath ); // TODO: Looks un-used. Remove?
 
 	/**
-	 * Create a {@link ReadData} through which data at the normal key can be read.
+	 * Create a {@link VolatileReadData} through which data at the normal key
+	 * can be read.
 	 * <p>
-	 * Implementations should read lazily if possible. Consumers may call {@link ReadData#materialize()} to force
-	 * a read operation if needed.
+	 * Implementations should read lazily if possible. Consumers may call {@link
+	 * ReadData#materialize()} to force a read operation if needed.
 	 * <p>
-	 * If supported by this KeyValueAccess implementation, partial reads are possible by calling slice on the output {@link ReadData}.
+	 * If supported by this KeyValueAccess implementation, partial reads are
+	 * possible by {@link ReadData#slice slicing} the returned {@code ReadData}.
+	 * <p>
+	 * The resulting {@code VolatileReadData} is potentially lazy. If the requested
+	 * key does not exist, it will throw {@code N5NoSuchKeyException}. Whether
+	 * the exception is thrown when {@link KeyValueAccess#createReadData(String)}] is called,
+	 * or when trying to materialize the {@code VolatileReadData} is implementation dependent.
+	 *
+	 * @param normalPath
+	 * 		is expected to be in normalized form, no further efforts are made to normalize it
+	 *
+	 * @return a ReadData
+	 *
+	 * @throws N5IOException
+	 * 		if an error occurs
+	 */
+	VolatileReadData createReadData(final String normalPath) throws N5IOException;
+
+	/**
+	 * Write {@code data} to the given {@code normalPath}.
+	 * <p>
+	 * Existing data at {@code normalPath} will be overridden.
+	 *
+	 * @param normalPath
+	 * 		is expected to be in normalized form, no further efforts are made to normalize it
+	 * @param data
+	 * 		the data to write
 	 *
-	 * @param normalPath is expected to be in normalized form, no further efforts are made to normalize it
-	 * @return a materialized Read data
-	 * @throws N5IOException if an error occurs
+	 * @throws N5IOException
+	 * 		if an error occurs
 	 */
-	ReadData createReadData( final String normalPath ) throws N5IOException;
+	void write(String normalPath, ReadData data) throws N5IOException;
 
 	/**
 	 * Create a lock on a path for reading. This isn't meant to be kept
@@ -268,14 +294,15 @@ default URI uri(final String uriString) throws URISyntaxException {
 	 * @return the locked channel
 	 * @throws N5IOException
 	 *             if a locked channel could not be created
+	 * @deprecated migrate to {@link KeyValueAccess#createReadData(String)}
 	 */
+	@Deprecated
 	LockedChannel lockForReading( final String normalPath ) throws N5IOException;
 
 	/**
-	 * Create an exclusive lock on a path for writing. If the file doesn't
-	 * exist yet, it will be created, including all directories leading up to
-	 * it. This lock isn't meant to be kept around. Create, use, [auto]close,
-	 * e.g.
+	 * Create an exclusive lock on a path for writing. If the file doesn't exist
+	 * yet, it will be created, including all directories leading up to it. This
+	 * lock isn't meant to be kept around. Create, use, [auto]close, e.g.
 	 * <code>
 	 * try (final lock = store.lockForWriting()) {
 	 *   ...
@@ -288,7 +315,9 @@ default URI uri(final String uriString) throws URISyntaxException {
 	 * @return the locked channel
 	 * @throws N5IOException
 	 *             if a locked channel could not be created
+	 * @deprecated migrate to {@link KeyValueAccess#write(String, ReadData)}
 	 */
+	@Deprecated
 	LockedChannel lockForWriting( final String normalPath ) throws N5IOException;
 
 	/**
@@ -338,47 +367,4 @@ default URI uri(final String uriString) throws URISyntaxException {
 	 *            if an error occurs during deletion
 	 */
 	void delete( final String normalPath ) throws N5IOException;
-
-	/**
-	 * A lazy reading strategy for lazy, partial reading of data from some source.
-	 * <p>
-	 * Implementations of this interface handle the specifics of accessing data from
-	 * their respective sources.
-	 * 
-	 * @see ReadData
-	 * @see KeyValueAccessReadData
-	 */
-	interface LazyRead {
-
-		/**
-		 * Materializes a portion of the data into a concrete {@link ReadData}
-		 * instance.
-		 * <p>
-		 * This method performs the actual read operation from the underlying
-		 * source, loading only the requested portion of data. The implementation
-		 * should handle bounds checking and throw appropriate exceptions for
-		 * invalid ranges.
-		 *
-		 * @param offset
-		 *            the starting position in the data source
-		 * @param length
-		 *            the number of bytes to read, or -1 to read from offset to end
-		 * @return a materialized {@link ReadData} instance containing the requested
-		 *         data
-		 * @throws N5IOException
-		 *             if any I/O error occurs
-		 */
-		ReadData materialize(long offset, long length) throws N5IOException;
-
-		/**
-		 * Returns the total size of the data source in bytes.
-		 * 
-		 * @return the size of the data source in bytes
-		 * @throws N5IOException
-		 * 		if an I/O error occurs while trying to get the length
-		 */
-		long size() throws N5IOException;
-
-	}
-
 }