HDDS-15424. Fix Concurrent positional read

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

@@ -23,6 +23,7 @@
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -76,7 +77,7 @@ public class BlockInputStream extends BlockExtendedInputStream {
   private XceiverClientShortCircuit xceiverClientShortCircuit;
   private final AtomicBoolean fallbackToGrpc = new AtomicBoolean(false);
   private volatile FileInputStream blockFileInputStream;
-  private boolean initialized = false;
+  private volatile boolean initialized = false;
   // TODO: do we need to change retrypolicy based on exception.
   private final RetryPolicy retryPolicy;
 
@@ -486,6 +487,106 @@ protected synchronized int readWithStrategy(ByteReaderStrategy strategy)
    *    2. chunkStream[2] will be seeked to position 10
    *       (= 90 - chunkOffset[2] (= 80)).
    */
+  /**
+   * Stateless positioned read across this block's chunks. Fills up to
+   * {@code dst.remaining()} bytes starting from {@code blockRelativePosition}
+   * without mutating this stream's cursor ({@code chunkIndex},
+   * {@code blockPosition}) or the sequential chunk streams' buffered state.
+   * Each covering chunk is read through an ephemeral {@link ChunkInputStream}
+   * closed as soon as its bytes have been copied.
+   *
+   * @return bytes copied into {@code dst}, or {@link #EOF} at EOF
+   */
+  int readPositioned(long blockRelativePosition, ByteBuffer dst)
+      throws IOException {
+    if (!initialized) {
+      initialize();
+    }
+    final long[] offsets = chunkOffsets;
+    final BlockData currentBlockData = blockData;
+    final long blockLength = length;
+    if (offsets == null || currentBlockData == null
+        || blockRelativePosition < 0 || blockRelativePosition >= blockLength) {
+      return EOF;
+    }
+
+    final List<ChunkInfo> chunkInfos = currentBlockData.getChunksList();
+    int index = Arrays.binarySearch(offsets, blockRelativePosition);
+    if (index < 0) {
+      index = -index - 2;
+    }
+
+    long pos = blockRelativePosition;
+    int totalReadLen = 0;
+    while (dst.hasRemaining() && pos < blockLength && index < chunkInfos.size()) {
+      final ChunkInfo chunkInfo = chunkInfos.get(index);
+      final long chunkOffset = pos - offsets[index];
+      final long numBytesToRead = Math.min(
+          Math.min(dst.remaining(), chunkInfo.getLen() - chunkOffset), blockLength - pos);
+      if (numBytesToRead <= 0) {
+        index++;
+        continue;
+      }
+      final int numBytesRead =
+          readChunkAt(chunkInfo, chunkOffset, (int) numBytesToRead, dst);
+      totalReadLen += numBytesRead;
+      pos += numBytesRead;
+      index++;
+    }
+    return totalReadLen == 0 ? EOF : totalReadLen;
+  }
+
+  /**
+   * Read {@code numBytesToRead} bytes starting at {@code chunkOffset} of the given chunk into {@code dst}
+   * through an ephemeral {@link ChunkInputStream}, retrying like {@link #readWithStrategy(ByteReaderStrategy)}
+   * but with a retry counter local to this call.
+   */
+  private int readChunkAt(ChunkInfo chunkInfo, long chunkOffset, int numBytesToRead, ByteBuffer dst)
+      throws IOException {
+    final int startPosition = dst.position();
+    int preadRetries = 0;
+    while (true) {
+      final ChunkInputStream chunkStream = createChunkInputStream(chunkInfo);
+      final int numBytesRead;
+      try {
+        final int oldLimit = dst.limit();
+        try {
+          dst.limit(startPosition + numBytesToRead);
+          numBytesRead = chunkStream.readPositioned(chunkOffset, dst);
+        } finally {
+          dst.limit(oldLimit);
+        }
+      } catch (SCMSecurityException ex) {
+        throw ex;
+      } catch (StorageContainerException ex) {
+        if (!shouldRetryRead(ex, retryPolicy, ++preadRetries)) {
+          throw ex;
+        }
+        handleReadError(ex);
+        dst.position(startPosition);
+        continue;
+      } catch (IOException ex) {
+        if (!shouldRetryRead(ex, retryPolicy, ++preadRetries)) {
+          throw ex;
+        }
+        if (isConnectivityIssue(ex)) {
+          handleReadError(ex);
+        }
+        dst.position(startPosition);
+        continue;
+      } finally {
+        chunkStream.close();
+      }
+
+      if (numBytesRead != numBytesToRead) {
+        throw new IOException(String.format(
+            "Inconsistent read for chunkName=%s length=%d numBytesToRead=%d numBytesRead=%d",
+            chunkInfo.getChunkName(), chunkInfo.getLen(), numBytesToRead, numBytesRead));
+      }
+      return numBytesRead;
+    }
+  }
+
   @Override
   public synchronized void seek(long pos) throws IOException {
     if (!initialized) {
@@ -638,7 +739,7 @@ private synchronized void storePosition() {
     blockPosition = getPos();
   }
 
-  private void handleReadError(IOException cause) throws IOException {
+  private synchronized void handleReadError(IOException cause) throws IOException {
     releaseClient();
     final List<ChunkInputStream> inputStreams = this.chunkStreams;
     if (inputStreams != null) {

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

@@ -429,6 +429,109 @@ protected void readChunkDataIntoBuffers(ChunkInfo readChunkInfo)
     allocated = true;
   }
 
+  /**
+   * Whether this chunk stream can serve positioned reads without holding a
+   * lock. A plain chunk read is a self-contained RPC, so concurrent callers
+   * reading different ranges do not interfere. Overridden by
+   * {@link LocalChunkInputStream} uses positional {@link FileChannel} reads on
+   * the shared block channel, so concurrent callers on different chunks do not
+   * interfere.
+   */
+  boolean supportsConcurrentPositionedRead() {
+    return true;
+  }
+
+  /**
+   * Stateless positioned read of up to {@code dst.remaining()} bytes starting
+   * at {@code chunkRelativePosition} within this chunk. Unlike the buffered
+   * {@link #read} path, this does not read or mutate any of the instance's
+   * buffer/position state ({@code buffers}, {@code chunkPosition},
+   * {@code bufferOffsetWrtChunkData}, ...), so it is safe to call concurrently
+   * from multiple threads sharing the same stream.
+   *
+   * @param chunkRelativePosition start offset within this chunk
+   * @param dst destination buffer
+   * @return number of bytes copied into {@code dst}, or {@link #EOF} at EOF
+   */
+  int readPositioned(long chunkRelativePosition, ByteBuffer dst)
+      throws IOException {
+    if (supportsConcurrentPositionedRead()) {
+      return doPositionedRead(chunkRelativePosition, dst);
+    }
+    // Local (short-circuit) reads share a FileChannel cursor; serialize them.
+    synchronized (this) {
+      return doPositionedRead(chunkRelativePosition, dst);
+    }
+  }
+
+  private int doPositionedRead(long chunkRelativePosition, ByteBuffer dst)
+      throws IOException {
+    if (chunkRelativePosition < 0 || chunkRelativePosition >= length) {
+      return EOF;
+    }
+    final int toRead =
+        (int) Math.min(dst.remaining(), length - chunkRelativePosition);
+    if (toRead == 0) {
+      return 0;
+    }
+
+    acquireClient();
+
+    final long adjustedOffset;
+    final long adjustedLen;
+    if (verifyChecksum) {
+      Pair<Long, Long> boundaries =
+          computeChecksumBoundaries(chunkRelativePosition, toRead);
+      adjustedOffset = boundaries.getLeft();
+      adjustedLen = boundaries.getRight();
+    } else {
+      adjustedOffset = chunkRelativePosition;
+      adjustedLen = toRead;
+    }
+
+    final ChunkInfo readChunkInfo = ChunkInfo.newBuilder(chunkInfo)
+        .setOffset(chunkInfo.getOffset() + adjustedOffset)
+        .setLen(adjustedLen)
+        .build();
+
+    final ByteBuffer[] readBuffers = readChunk(readChunkInfo);
+    return copyRange(readBuffers, chunkRelativePosition - adjustedOffset,
+        toRead, dst);
+  }
+
+  /**
+   * Copy {@code toCopy} bytes from {@code src} buffers, skipping the first
+   * {@code skip} bytes, into {@code dst}. Operates on duplicates so the source
+   * buffers' positions are left untouched.
+   */
+  private static int copyRange(ByteBuffer[] src, long skip, int toCopy,
+      ByteBuffer dst) {
+    long remainingSkip = skip;
+    int copied = 0;
+    for (ByteBuffer buffer : src) {
+      if (copied >= toCopy) {
+        break;
+      }
+      ByteBuffer dup = buffer.duplicate();
+      if (remainingSkip > 0) {
+        int skipHere = (int) Math.min(remainingSkip, dup.remaining());
+        dup.position(dup.position() + skipHere);
+        remainingSkip -= skipHere;
+        if (!dup.hasRemaining()) {
+          continue;
+        }
+      }
+      int n = Math.min(dup.remaining(), toCopy - copied);
+      if (n <= 0) {
+        continue;
+      }
+      dup.limit(dup.position() + n);
+      dst.put(dup);
+      copied += n;
+    }
+    return copied;
+  }
+
   /**
    * Send RPC call to get the chunk from the container.
    */

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

@@ -72,6 +72,11 @@ public class LocalChunkInputStream extends ChunkInputStream
     }
   }
 
+  @Override
+  boolean supportsConcurrentPositionedRead() {
+    return true;
+  }
+
   /**
    * Get the chunk from the local block replica.
    */
@@ -82,12 +87,34 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
     int bytesPerChecksum = chunkInfo.getChecksumData().getBytesPerChecksum();
     final ByteBuffer[] buffers = BufferUtils.assignByteBuffers(readChunkInfo.getLen(),
         bytesPerChecksum);
-    dataIn.position(readChunkInfo.getOffset()).read(buffers);
+    readAtOffset(buffers, readChunkInfo.getOffset());
     Arrays.stream(buffers).forEach(ByteBuffer::flip);
     validator.accept(Arrays.asList(buffers), readChunkInfo);
     return buffers;
   }
 
+  /**
+   * Read into {@code buffers} starting at {@code fileOffset} using positional
+   * {@link FileChannel} reads so concurrent callers on different chunks (which
+   * share the same underlying channel) do not stomp each other's cursor.
+   */
+  private void readAtOffset(ByteBuffer[] buffers, long fileOffset) throws IOException {
+    long pos = fileOffset;
+    for (ByteBuffer buffer : buffers) {
+      while (buffer.hasRemaining()) {
+        int n = dataIn.read(buffer, pos);
+        if (n <= 0) {
+          if (buffer.hasRemaining()) {
+            throw new IOException("Failed to read chunk data at offset " + pos
+                + " for block chunk " + chunkInfo.getChunkName());
+          }
+          break;
+        }
+        pos += n;
+      }
+    }
+  }
+
   private void validateChunk(List<ByteBuffer> bufferList, ChunkInfo readChunkInfo)
       throws OzoneChecksumException {
     if (verifyChecksum) {