HDDS-16396. Support concurrent positioned reads for EC and encrypted keys

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java

@@ -476,7 +476,7 @@ protected synchronized int readWithStrategy(ByteReaderStrategy strategy)
   @Override
   public int readPositioned(long blockRelativePosition, ByteBuffer dst) throws IOException {
     if (!dst.hasRemaining()) {
-      return EOF;
+      return 0;
     }
     if (blockRelativePosition < 0) {
       return EOF;

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java

@@ -446,13 +446,13 @@ protected void readChunkDataIntoBuffers(ChunkInfo readChunkInfo)
    * @return number of bytes copied into {@code dst}, or {@link #EOF} at EOF
    */
   int readPositioned(long chunkRelativePosition, ByteBuffer dst) throws IOException {
+    if (!dst.hasRemaining()) {
+      return 0;
+    }
     if (chunkRelativePosition < 0 || chunkRelativePosition >= length) {
       return EOF;
     }
     final int toRead = (int) Math.min(dst.remaining(), length - chunkRelativePosition);
-    if (toRead == 0) {
-      return 0;
-    }
 
     final ChunkInfo readChunkInfo = getChunkInfo(chunkRelativePosition, toRead);
     final long adjustedOffset = readChunkInfo.getOffset() - chunkInfo.getOffset();

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java

@@ -17,12 +17,15 @@
 
 package org.apache.hadoop.hdds.scm.storage;
 
+import java.io.EOFException;
 import java.io.IOException;
-import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.nio.ReadOnlyBufferException;
 import org.apache.commons.lang3.NotImplementedException;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
 import org.apache.hadoop.fs.ByteBufferReadable;
 import org.apache.hadoop.fs.CanUnbuffer;
+import org.apache.hadoop.fs.FSInputStream;
 import org.apache.hadoop.fs.Seekable;
 import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.util.StringUtils;
@@ -31,20 +34,51 @@
  * Abstact class which extends InputStream and some common interfaces used by
  * various Ozone InputStream classes.
  */
-public abstract class ExtendedInputStream extends InputStream
-    implements Seekable, CanUnbuffer, ByteBufferReadable, StreamCapabilities {
+public abstract class ExtendedInputStream extends FSInputStream
+    implements Seekable, CanUnbuffer, ByteBufferReadable, ByteBufferPositionedReadable, StreamCapabilities {
 
   protected static final int EOF = -1;
 
+  @Override
+  public int read(long position, ByteBuffer buffer) throws IOException {
+    if (position < 0) {
+      throw new EOFException("position is negative: " + position);
+    }
+    if (buffer.isReadOnly()) {
+      throw new ReadOnlyBufferException();
+    }
+    return buffer.hasRemaining() ? readPositioned(position, buffer) : 0;
+  }
+
   /**
-   * Positioned read.
-   *
-   * @param position the starting position of the read.
-   * @param buffer the buffer for storing the data.
-   * @return true iff positioned read is supported in this implementation.
+   * Read without changing the sequential cursor. Implementations must allow concurrent calls.
    */
-  public boolean readFully(long position, ByteBuffer buffer) throws IOException {
-    return false;
+  protected abstract int readPositioned(long position, ByteBuffer buffer) throws IOException;
+
+  @Override
+  public void readFully(long position, ByteBuffer buffer) throws IOException {
+    do {
+      int n = read(position, buffer);
+      if (n < 0) {
+        throw new EOFException("End of stream at position " + position);
+      }
+      if (n == 0 && buffer.hasRemaining()) {
+        throw new IOException("No progress reading at position " + position);
+      }
+      position += n;
+    } while (buffer.hasRemaining());
+  }
+
+  @Override
+  public int read(long position, byte[] buffer, int offset, int length) throws IOException {
+    validatePositionedReadArgs(position, buffer, offset, length);
+    return read(position, ByteBuffer.wrap(buffer, offset, length));
+  }
+
+  @Override
+  public void readFully(long position, byte[] buffer, int offset, int length) throws IOException {
+    validatePositionedReadArgs(position, buffer, offset, length);
+    readFully(position, ByteBuffer.wrap(buffer, offset, length));
   }
 
   @Override
@@ -100,6 +134,7 @@ public synchronized boolean seekToNewSource(long l) throws IOException {
   @Override
   public boolean hasCapability(String capability) {
     switch (StringUtils.toLowerCase(capability)) {
+    case StreamCapabilities.PREADBYTEBUFFER:
     case StreamCapabilities.READBYTEBUFFER:
     case StreamCapabilities.UNBUFFER:
       return true;

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java

@@ -27,7 +27,6 @@
 import java.util.List;
 import java.util.Objects;
 import org.apache.hadoop.fs.FSExceptionMessages;
-import org.apache.ratis.util.Preconditions;
 
 /**
  * A stream for accessing multipart streams.
@@ -39,8 +38,6 @@ public class MultipartInputStream extends ExtendedInputStream {
 
   // List of PartInputStream, one for each part of the key
   private final List<? extends PartInputStream> partStreams;
-  private final boolean isStreamBlockInputStream;
-  private final boolean statelessPositionedReadSupported;
 
   // partOffsets[i] stores the index of the first data byte in
   // partStream w.r.t the whole key data.
@@ -50,7 +47,7 @@ public class MultipartInputStream extends ExtendedInputStream {
   // part[0]), partOffsets[1] = 200 and so on.
   private final long[] partOffsets;
 
-  private boolean closed;
+  private volatile boolean closed;
   // Index of the partStream corresponding to the current position of the
   // MultipartCryptoKeyInputStream.
   private int partIndex;
@@ -67,30 +64,18 @@ public MultipartInputStream(String keyName,
 
     this.key = keyName;
     this.partStreams = Collections.unmodifiableList(inputStreams);
-    this.isStreamBlockInputStream = !inputStreams.isEmpty() && inputStreams.get(0) instanceof StreamBlockInputStream;
 
     // Calculate and update the partOffsets
     this.partOffsets = new long[inputStreams.size()];
-    boolean statelessSupported = !inputStreams.isEmpty();
     int i = 0;
     long streamLength = 0L;
     for (PartInputStream partInputStream : inputStreams) {
       this.partOffsets[i++] = streamLength;
-      if (isStreamBlockInputStream) {
-        Preconditions.assertInstanceOf(partInputStream, StreamBlockInputStream.class);
-      } else if (statelessSupported && !(partInputStream instanceof BlockInputStream)) {
-        statelessSupported = false;
-      }
       streamLength += partInputStream.getLength();
     }
-    this.statelessPositionedReadSupported = statelessSupported;
     this.length = streamLength;
   }
 
-  public boolean isStreamBlockInputStream() {
-    return isStreamBlockInputStream;
-  }
-
   @Override
   protected synchronized int readWithStrategy(ByteReaderStrategy strategy)
       throws IOException {
@@ -195,45 +180,28 @@ public synchronized void seek(long pos) throws IOException {
   }
 
   @Override
-  public boolean readFully(long position, ByteBuffer buffer) throws IOException {
-    if (!statelessPositionedReadSupported) {
-      return false;
-    }
-    return readFullyStateless(position, buffer);
-  }
-
-  /**
-   * Shared stateless positioned-read loop across parts using {@link PartInputStream#readPositioned}.
-   */
-  private boolean readFullyStateless(long position, ByteBuffer buffer) throws IOException {
+  protected int readPositioned(long position, ByteBuffer buffer) throws IOException {
+    checkOpen();
     if (!buffer.hasRemaining()) {
-      return true;
+      return 0;
     }
-
     long pos = position;
     int bytesRead = 0;
-    while (buffer.hasRemaining()) {
-      if (pos < 0 || pos >= length) {
-        if (bytesRead > 0) {
-          return true;
-        }
-        throw new EOFException("EOF encountered at pos: " + pos +
-            " for key: " + key);
+    while (buffer.hasRemaining() && pos < length) {
+      int index = binarySearchOffsetIndex(partOffsets, pos);
+      PartInputStream part = partStreams.get(index);
+      long partPosition = pos - partOffsets[index];
+      final int n = part.read(partPosition, buffer);
+      if (n < 0) {
+        break;
       }
-      int idx = binarySearchOffsetIndex(partOffsets, pos);
-      long partPos = pos - partOffsets[idx];
-      int n = partStreams.get(idx).readPositioned(partPos, buffer);
-      if (n <= 0) {
-        if (bytesRead > 0) {
-          return true;
-        }
-        throw new EOFException("EOF encountered at pos: " + pos +
-            " for key: " + key);
+      if (n == 0) {
+        throw new IOException("No progress reading key " + key + " at position " + pos);
       }
       bytesRead += n;
       pos += n;
     }
-    return true;
+    return bytesRead == 0 ? EOF : bytesRead;
   }
 
   static int binarySearchOffsetIndex(long[] offsets, long pos) {

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/PartInputStream.java

@@ -18,32 +18,21 @@
 package org.apache.hadoop.hdds.scm.storage;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
 import org.apache.hadoop.fs.CanUnbuffer;
 import org.apache.hadoop.fs.Seekable;
 
 /**
  * A stream that can be a part of a {@link MultipartInputStream}.
+ * Positioned reads must allow concurrent calls without changing the sequential cursor.
  */
 public interface PartInputStream
-    extends CanUnbuffer, Seekable {
+    extends CanUnbuffer, Seekable, ByteBufferPositionedReadable {
   long getLength();
 
   default long getRemaining() throws IOException {
     return getLength() - getPos();
   }
 
   void close() throws IOException;
-
-  /**
-   * Positioned read within this part starting at {@code partOffset}.
-   *
-   * @return bytes copied into {@code buffer}, {@code -1} if {@code buffer} has no remaining space
-   *         or at EOF
-   * @throws UnsupportedOperationException if this part stream does not support positioned read
-   */
-  default int readPositioned(long partOffset, ByteBuffer buffer) throws IOException {
-    throw new UnsupportedOperationException(
-        "Positioned read is not supported by " + getClass().getName());
-  }
 }

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java

@@ -185,12 +185,12 @@ synchronized int readFully(ByteBuffer targetBuf, boolean preRead) throws IOExcep
    *
    * @param blockOffset the offset within the block to read from.
    * @param dst the buffer to read into.
-   * @return the number of bytes copied into {@code dst}, or -1 if no byte could be read.
+   * @return the number of bytes copied into {@code dst}, or -1 at EOF.
    */
   @Override
   public int readPositioned(long blockOffset, ByteBuffer dst) throws IOException {
     if (!dst.hasRemaining()) {
-      return EOF;
+      return 0;
     }
     if (blockOffset < 0 || blockOffset >= blockLength) {
       return EOF;

hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java

@@ -31,6 +31,7 @@
 import java.util.TreeMap;
 import java.util.function.Function;
 import org.apache.hadoop.fs.FSExceptionMessages;
+import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.client.ContainerBlockID;
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
@@ -300,6 +301,17 @@ protected long remaining() {
     return blockLength() - position;
   }
 
+  @Override
+  protected int readPositioned(long offset, ByteBuffer buffer) {
+    throw new UnsupportedOperationException("Use ECBlockInputStreamProxy for positioned reads");
+  }
+
+  @Override
+  public boolean hasCapability(String capability) {
+    return !StreamCapabilities.PREADBYTEBUFFER.equalsIgnoreCase(capability)
+        && super.hasCapability(capability);
+  }
+
   @Override
   public synchronized int read(byte[] b, int off, int len) throws IOException {
     return read(ByteBuffer.wrap(b, off, len));

hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java

@@ -19,10 +19,10 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Function;
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
@@ -56,8 +56,8 @@ public class ECBlockInputStreamProxy extends BlockExtendedInputStream {
 
   private BlockExtendedInputStream blockReader;
   private boolean reconstructionReader = false;
-  private List<DatanodeDetails> failedLocations = new ArrayList<>();
-  private boolean closed = false;
+  private final CopyOnWriteArrayList<DatanodeDetails> failedLocations;
+  private volatile boolean closed = false;
   private OzoneClientConfig config;
 
   /**
@@ -102,12 +102,21 @@ public ECBlockInputStreamProxy(ECReplicationConfig repConfig,
       BlockLocationInfo> refreshFunction,
       ECBlockInputStreamFactory streamFactory,
       OzoneClientConfig config) {
+    this(repConfig, blockInfo, xceiverClientFactory, refreshFunction, streamFactory, config,
+        new CopyOnWriteArrayList<>());
+  }
+
+  private ECBlockInputStreamProxy(ECReplicationConfig repConfig, BlockLocationInfo blockInfo,
+      XceiverClientFactory xceiverClientFactory, Function<BlockID, BlockLocationInfo> refreshFunction,
+      ECBlockInputStreamFactory streamFactory, OzoneClientConfig config,
+      CopyOnWriteArrayList<DatanodeDetails> failedLocations) {
     this.repConfig = repConfig;
     this.blockInfo = blockInfo;
     this.ecBlockInputStreamFactory = streamFactory;
     this.xceiverClientFactory = xceiverClientFactory;
     this.refreshFunction = refreshFunction;
     this.config = config;
+    this.failedLocations = failedLocations;
 
     setReaderType();
     createBlockReader();
@@ -116,9 +125,10 @@ public ECBlockInputStreamProxy(ECReplicationConfig repConfig,
   private synchronized void setReaderType() {
     int expected = expectedDataLocations(repConfig, getLength());
     int available = availableDataLocations(blockInfo.getPipeline(), expected);
-    reconstructionReader = available < expected;
+    reconstructionReader = available < expected || !failedLocations.isEmpty();
     if (reconstructionReader) {
-      LOG.info("Data locations available: {} < expected: {}, using reconstruction read", available, expected);
+      LOG.info("Data locations available: {}, expected: {}, failed: {}, using reconstruction read",
+          available, expected, failedLocations.size());
     }
   }
 
@@ -197,11 +207,28 @@ public synchronized int read(ByteBuffer buf) throws IOException {
     return totalRead;
   }
 
+  @Override
+  protected int readPositioned(long position, ByteBuffer buffer) throws IOException {
+    ensureNotClosed();
+    if (!buffer.hasRemaining()) {
+      return 0;
+    }
+    if (position >= blockInfo.getLength()) {
+      return EOF;
+    }
+    // Each request owns its cursor and reconstruction buffers; failed locations are shared.
+    try (ECBlockInputStreamProxy reader = new ECBlockInputStreamProxy(repConfig, blockInfo,
+        xceiverClientFactory, refreshFunction, ecBlockInputStreamFactory, config, failedLocations)) {
+      reader.seek(position);
+      return reader.read(buffer);
+    }
+  }
+
   private synchronized void failoverToReconstructionRead(
       List<DatanodeDetails> badLocations, long lastPosition)
       throws IOException {
     if (badLocations != null) {
-      failedLocations.addAll(badLocations);
+      failedLocations.addAllAbsent(badLocations);
     }
     blockReader.close();
     reconstructionReader = true;