4.0.0-alpha-12 candidate

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

@@ -1,6 +1,8 @@
 package org.janelia.saalfeldlab.n5;
 
+import org.apache.commons.io.input.ProxyInputStream;
 import org.janelia.saalfeldlab.n5.readdata.ReadData;
+import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;
 
 import java.io.*;
 
@@ -23,7 +25,15 @@ public Reader newReader() throws N5Exception.N5IOException {
 
     @Override
     public InputStream newInputStream() throws N5Exception.N5IOException {
-        return kva.createReadData(key).inputStream();
+
+        VolatileReadData volatileReadData = kva.createReadData(key);
+        return new ProxyInputStream(volatileReadData.inputStream()) {
+            @Override
+            public void close() throws IOException {
+                super.close();
+                volatileReadData.close();
+            }
+        };
     }
 
     @Override

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

@@ -2,12 +2,11 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.nio.channels.FileChannel;
 import java.nio.channels.FileLock;
 import java.nio.channels.OverlappingFileLockException;
-import java.nio.file.Files;
 import java.nio.file.Path;
-import java.nio.file.StandardOpenOption;
 
 /**
  * Holds a channel and system-level file lock (shared for writing, non-shared
@@ -17,11 +16,14 @@
 class ChannelLock implements Closeable {
 
 	private final FileChannel channel;
-	private final FileLock lock;
+
+	//Used to ensure a hard reference exists until we close
+	@SuppressWarnings({"unused", "FieldCanBeLocal"})
+    private final FileLock unusedHardRef;
 
 	private ChannelLock(final FileChannel channel, final FileLock lock) {
 		this.channel = channel;
-		this.lock = lock;
+		this.unusedHardRef = lock;
 	}
 
 	public void close() throws IOException {
@@ -55,7 +57,7 @@ FileChannel getChannel() {
 	 */
 	static ChannelLock lock(final Path path, final boolean forWriting) throws IOException {
 
-		final FileChannel channel = openFileChannel(path, forWriting);
+		final FileChannel channel = FsIoPolicy.openFileChannel(path, forWriting);
 		try {
 			while (true) {
 				try {
@@ -93,7 +95,7 @@ static ChannelLock tryLock(final Path path, final boolean forWriting) throws IOE
 
 		FileChannel channel = null;
 		try {
-			channel = openFileChannel(path, forWriting);
+			channel = FsIoPolicy.openFileChannel(path, forWriting);
 			final FileLock lock = channel.tryLock(0, Long.MAX_VALUE, !forWriting);
 			return lock == null ? null : new ChannelLock(channel, lock);
 		} catch (Exception e) {
@@ -102,32 +104,12 @@ static ChannelLock tryLock(final Path path, final boolean forWriting) throws IOE
 		}
 	}
 
-	/**
-	 * Opens a file channel. If the channel is opened {@code forWriting},
-	 * then this may create the file and the parent directories as needed.
-	 *
-	 * @throws IOException
-	 * 		if the channel cannot be opened
-	 */
-	private static FileChannel openFileChannel(final Path path, final boolean forWriting) throws IOException {
-
-		if (forWriting) {
-			final Path parent = path.getParent();
-			if (parent != null) {
-				Files.createDirectories(parent);
-			}
-			return FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.CREATE);
-		} else {
-			return FileChannel.open(path, StandardOpenOption.READ);
-		}
-	}
-
 	private static void closeQuietly(final FileChannel fileChannel) {
 		if (fileChannel != null) {
 			try {
 				fileChannel.close();
-			} catch (final IOException ignored) {
+			} catch (final IOException | UncheckedIOException ignored) {
 			}
-		}
+        }
 	}
 }

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

@@ -37,6 +37,7 @@
 import com.google.gson.JsonPrimitive;
 import com.google.gson.JsonSerializationContext;
 import com.google.gson.JsonSerializer;
+
 /**
  * Enumerates available data types.
  *

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

@@ -89,13 +89,13 @@ public class DatasetAttributes implements Serializable {
 
 	private final long[] dimensions;
 
+	// number of samples per chunk per dimension
+	private final int[] chunkSize;
+
 	// number of samples per block per dimension
+	// identical to chunkSize for non-sharded datasets
 	private final int[] blockSize;
 
-	// TODO add a getter?
-	// the shard size
-	private final int[] outerBlockSize;
-
 	private final DataType dataType;
 
 	private final JsonElement defaultValue;
@@ -108,7 +108,7 @@ public class DatasetAttributes implements Serializable {
 
 	public DatasetAttributes(
 			final long[] dimensions,
-			final int[] outerBlockSize,
+			final int[] blockSize,
 			final DataType dataType,
 			final JsonElement defaultValue,
 			final BlockCodecInfo blockCodecInfo,
@@ -117,7 +117,7 @@ public DatasetAttributes(
 
 		this.dimensions = dimensions;
 		this.dataType = dataType;
-		this.outerBlockSize = outerBlockSize;
+		this.blockSize = blockSize;
 		this.defaultValue = defaultValue == null ? JsonNull.INSTANCE : defaultValue;
 
 		this.blockCodecInfo = blockCodecInfo == null ? defaultBlockCodecInfo() : blockCodecInfo;
@@ -131,7 +131,7 @@ public DatasetAttributes(
 					.toArray(DataCodecInfo[]::new);
 
 		access = createDatasetAccess();
-		blockSize = access.getGrid().getBlockSize(0);
+		chunkSize = access.getGrid().getBlockSize(0);
 	}
 
 	public DatasetAttributes(
@@ -189,7 +189,7 @@ public DatasetAttributes(
 		this(dimensions, blockSize, dataType, new DataCodecInfo[0]);
 	}
 
-	protected DatasetAccess<?> createDatasetAccess() {
+	private DatasetAccess<?> createDatasetAccess() {
 
 		final int m = nestingDepth(blockCodecInfo);
 
@@ -200,7 +200,7 @@ protected DatasetAccess<?> createDatasetAccess() {
 		// NestedGrid validates block sizes, so instantiate it before creating the blockCodecs
 		// blockCodecInfo.create below could fail unexpecedly with invalid
 		// blockSizes so validate first
-		blockSizes[m - 1] = outerBlockSize;
+		blockSizes[m - 1] = blockSize;
 		BlockCodecInfo tmpInfo = blockCodecInfo;
 		for (int l = m - 1; l > 0; --l) {
 			final ShardCodecInfo info = (ShardCodecInfo)tmpInfo;
@@ -210,7 +210,7 @@ protected DatasetAccess<?> createDatasetAccess() {
 
 		BlockCodecInfo currentBlockCodecInfo = blockCodecInfo;
 		DataCodecInfo[] currentDataCodecInfos = dataCodecInfos;
-		
+
 		DatasetCodecInfo[] datasetCodecInfos = this.datasetCodecInfos;
 
 		final NestedGrid grid = new NestedGrid(blockSizes, dimensions);
@@ -274,6 +274,11 @@ public int getNumDimensions() {
 		return dimensions.length;
 	}
 
+	public int[] getChunkSize() {
+
+		return chunkSize;
+	}
+
 	public int[] getBlockSize() {
 
 		return blockSize;
@@ -318,9 +323,9 @@ public DataType getDataType() {
 	 *
 	 * @return the {@code DatasetAccess} for this dataset
 	 */
-	<T> DatasetAccess<T> getDatasetAccess() {
+	protected <T> DatasetAccess<T> getDatasetAccess() {
 
-		return (DatasetAccess<T>)access;
+		return (DatasetAccess<T>) access;
 	}
 
 	/**
@@ -334,7 +339,6 @@ public NestedGrid getNestedBlockGrid() {
 		return getDatasetAccess().getGrid();
 	}
 
-
 	public BlockCodecInfo getBlockCodecInfo() {
 
 		return blockCodecInfo;
@@ -359,7 +363,7 @@ public HashMap<String, Object> asMap() {
 
 		final HashMap<String, Object> map = new HashMap<>();
 		map.put(DIMENSIONS_KEY, dimensions);
-		map.put(BLOCK_SIZE_KEY, blockSize);
+		map.put(BLOCK_SIZE_KEY, chunkSize);
 		map.put(DATA_TYPE_KEY, dataType);
 		map.put(COMPRESSION_KEY, getCompression());
 		return map;
@@ -425,7 +429,7 @@ public static class DatasetAttributesAdapter implements JsonSerializer<DatasetAt
 
 			final JsonObject obj = new JsonObject();
 			obj.add(DIMENSIONS_KEY, context.serialize(src.dimensions));
-			obj.add(BLOCK_SIZE_KEY, context.serialize(src.blockSize));
+			obj.add(BLOCK_SIZE_KEY, context.serialize(src.chunkSize));
 			obj.add(DATA_TYPE_KEY, context.serialize(src.dataType));
 
 			final DataCodecInfo[] codecs = src.dataCodecInfos;

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

@@ -6,8 +6,10 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.io.OutputStream;
 import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
 import java.nio.channels.FileChannel;
 import java.nio.file.*;
 
@@ -29,11 +31,65 @@ else if (length >= 0 && offset + length > channelSize)
         return true;
     }
 
+    /**
+     * Opens a file channel. If the channel is opened {@code forWriting},
+     * then this may create the file and the parent directories as needed.
+     *
+     * @throws IOException
+     * 		if the channel cannot be opened
+     */
+    static FileChannel openFileChannel(final Path path, final boolean forWriting) throws IOException {
+
+        if (forWriting) {
+            final Path parent = path.getParent();
+            /* if not null and not directory, it will call `createDirectories` but we expect it to throw an IOException */
+            if (parent != null && !parent.toFile().isDirectory()) {
+                Files.createDirectories(parent);
+            }
+            return FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+        } else {
+            return FileChannel.open(path, StandardOpenOption.READ);
+        }
+    }
+
+
+    /**
+     * This method is necessary to handle the situtation where writing is successful, but `close` fails on the file channel.
+     * This has been observed to happen fairly consistently on MacOS when writing to a file mounted over SMB.
+     *
+     * @param readData to write to the {@code Path}
+     * @param path to write to
+     * @throws IOException if writing failed.
+     */
+    private static void writeToPathIgnoreCloseException(ReadData readData, Path path) throws IOException {
+
+        FileChannel channel = openFileChannel(path, true);
+        OutputStream os = Channels.newOutputStream(channel);
+
+        try {
+            readData.writeTo(os);
+            os.flush();
+            channel.force(true);
+        } catch (Throwable e) {
+            os.close();
+            channel.close();
+            throw e;
+        }
+
+        /* if we get here, the write succeeded, and the os/channel may not be closed yet */
+        try {
+            os.close();
+            channel.close();
+        } catch (IOException | UncheckedIOException ignore) {
+            /* Ignore; we know the data was written already. */
+        }
+    }
+
     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);
+            writeToPathIgnoreCloseException(readData, path);
         }
 
         @Override
@@ -69,6 +125,8 @@ public VolatileReadData read(String key) throws IOException {
         @Override
         public void delete(final String key) throws IOException {
             final Path path = Paths.get(key);
+            if (!Files.isRegularFile(path))
+                Files.delete(path);
             try (LockedFileChannel ignore = FILE_LOCK_MANAGER.lockForWriting(path)) {
                 Files.delete(path);
             }
@@ -110,6 +168,7 @@ public ReadData materialize(final long offset, final long length) {
                 throw new N5Exception.N5IOException("FileLazyRead is already closed.");
             }
 
+            ReadData readData = null;
             try (final FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) {
 
                 channel.position(offset);
@@ -127,13 +186,19 @@ public ReadData materialize(final long offset, final long length) {
                 final byte[] data = new byte[(int) size];
                 final ByteBuffer buf = ByteBuffer.wrap(data);
                 channel.read(buf);
-                return ReadData.from(data);
+                readData = ReadData.from(data);
 
             } catch (final NoSuchFileException e) {
                 throw new N5Exception.N5NoSuchKeyException("No such file", e);
             } catch (IOException | UncheckedIOException e) {
-                throw new N5Exception.N5IOException(e);
+                /* Occasionally (frequently for some source remote mounted file systems) this can throw exceptions during
+                 * `channel.close()` which is called automatically in the try-with-resources block. In this case, we have
+                 * successfully read the data, and we can return it, and ignore the exception.
+                 * */
+                if (readData == null)
+                    throw new N5Exception.N5IOException(e);
             }
+            return readData;
         }
 
         @Override