HDDS-16325. Reuse chunk range readers for positioned reads

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

@@ -217,7 +217,7 @@ public synchronized void initialize() throws IOException {
     }
   }
 
-  private void refreshBlockInfo(IOException cause) throws IOException {
+  private synchronized void refreshBlockInfo(IOException cause) throws IOException {
     refreshBlockInfo(cause, blockID, pipelineRef, tokenRef, refreshFunction);
   }
 
@@ -383,8 +383,8 @@ protected synchronized void addStream(ChunkInfo chunkInfo) {
   protected ChunkInputStream createChunkInputStream(ChunkInfo chunkInfo) {
     if (blockFileInputStream != null) {
       // a non-empty blockFileInputStream means we have a direct local block replica to read from
-      return new LocalChunkInputStream(chunkInfo, blockID, xceiverClientFactory,
-          pipelineRef::get, verifyChecksum, tokenRef::get, xceiverClientShortCircuit, blockFileInputStream);
+      return new LocalChunkInputStream(chunkInfo, blockID, xceiverClientFactory, pipelineRef::get,
+          verifyChecksum, tokenRef::get, xceiverClientShortCircuit, blockFileInputStream.getChannel());
     } else {
       return new ChunkInputStream(chunkInfo, blockID,
           xceiverClientFactory, pipelineRef::get, verifyChecksum, tokenRef::get);
@@ -468,8 +468,7 @@ protected synchronized int readWithStrategy(ByteReaderStrategy strategy)
    * {@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.
+   * Reuses each chunk's range reader, whose buffers and client references belong to the request.
    *
    * @return bytes copied into {@code dst}, or {@link #EOF} at EOF
    */
@@ -486,30 +485,29 @@ public int readPositioned(long blockRelativePosition, ByteBuffer dst) throws IOE
       initialize();
     }
     final long[] offsets;
-    final BlockData currentBlockData;
+    final List<ChunkInputStream> streams;
     final long blockLength;
     synchronized (this) {
       checkOpen();
       if (chunkOffsets == null || blockData == null || blockRelativePosition >= length) {
         return EOF;
       }
 
-      offsets = Arrays.copyOf(chunkOffsets, chunkOffsets.length);
-      currentBlockData = blockData;
+      offsets = chunkOffsets;
+      streams = chunkStreams;
       blockLength = length;
     }
 
-    final List<ChunkInfo> chunkInfos = currentBlockData.getChunksList();
     int index = MultipartInputStream.binarySearchOffsetIndex(offsets, blockRelativePosition);
 
     long pos = blockRelativePosition;
     int totalReadLen = 0;
-    while (dst.hasRemaining() && pos < blockLength && index < chunkInfos.size()) {
-      final ChunkInfo chunkInfo = chunkInfos.get(index);
+    while (dst.hasRemaining() && pos < blockLength && index < streams.size()) {
+      final ChunkInputStream chunk = streams.get(index);
       final long chunkOffset = pos - offsets[index];
-      final long readLength = Math.min(Math.min(dst.remaining(), chunkInfo.getLen() - chunkOffset), blockLength - pos);
+      final long readLength = Math.min(Math.min(dst.remaining(), chunk.getLength() - chunkOffset), blockLength - pos);
       if (readLength > 0) {
-        final int numBytesRead = readChunkAt(chunkInfo, chunkOffset, (int) readLength, dst);
+        final int numBytesRead = readChunkAt(chunk, chunkOffset, (int) readLength, dst);
         totalReadLen += numBytesRead;
         pos += numBytesRead;
       }
@@ -518,56 +516,40 @@ public int readPositioned(long blockRelativePosition, ByteBuffer dst) throws IOE
     return totalReadLen == 0 ? EOF : totalReadLen;
   }
 
-  /**
-   * Read {@code readLength} 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 readLength, ByteBuffer dst) throws IOException {
-    final int startPosition = dst.position();
+  /** Read a chunk range with a request-local retry counter and buffer position. */
+  private int readChunkAt(ChunkInputStream chunk, long chunkOffset, int readLength, ByteBuffer dst) throws IOException {
     int preadRetries = 0;
     while (true) {
-      final ChunkInputStream chunkStream;
-      synchronized (this) {
-        checkOpen();
-        chunkStream = createChunkInputStream(chunkInfo);
-      }
+      checkOpen();
+      final ByteBuffer target = dst.duplicate();
+      target.limit(target.position() + readLength);
       final int numBytesRead;
       try {
-        final int oldLimit = dst.limit();
-        try {
-          dst.limit(startPosition + readLength);
-          numBytesRead = chunkStream.readPositioned(chunkOffset, dst);
-        } finally {
-          dst.limit(oldLimit);
-        }
+        numBytesRead = chunk.readPositioned(chunkOffset, target);
       } catch (SCMSecurityException ex) {
         throw ex;
       } catch (StorageContainerException ex) {
         if (!shouldRetryRead(ex, retryPolicy, ++preadRetries)) {
           throw ex;
         }
-        handleReadError(ex);
-        dst.position(startPosition);
+        refreshBlockInfo(ex);
         continue;
       } catch (IOException ex) {
         if (!shouldRetryRead(ex, retryPolicy, ++preadRetries)) {
           throw ex;
         }
         if (isConnectivityIssue(ex)) {
-          handleReadError(ex);
+          refreshBlockInfo(ex);
         }
-        dst.position(startPosition);
         continue;
-      } finally {
-        chunkStream.close();
       }
 
       if (numBytesRead != readLength) {
         throw new IOException(String.format(
             "Inconsistent read for chunkName=%s length=%d readLength=%d numBytesRead=%d",
-            chunkInfo.getChunkName(), chunkInfo.getLen(), readLength, numBytesRead));
+            chunk.getChunkName(), chunk.getLength(), readLength, numBytesRead));
       }
+      dst.position(target.position());
       return numBytesRead;
     }
   }

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

@@ -67,6 +67,7 @@ public class ChunkInputStream extends InputStream
   private XceiverClientSpi xceiverClient;
   private final Supplier<Pipeline> pipelineSupplier;
   private final boolean verifyChecksum;
+  private final ChunkReader reader;
   private boolean allocated = false;
   // Buffers to store the chunk data read from the DN container
   private ByteBuffer[] buffers;
@@ -114,6 +115,7 @@ public class ChunkInputStream extends InputStream
     this.xceiverClientFactory = xceiverClientFactory;
     this.pipelineSupplier = pipelineSupplier;
     this.verifyChecksum = verifyChecksum;
+    this.reader = new ChunkReader(chunkInfo, verifyChecksum, this::readChunkPositioned);
     this.tokenSupplier = tokenSupplier;
     validators = ContainerProtocolCalls.toValidatorList(this::validateChunk);
   }
@@ -293,30 +295,20 @@ protected synchronized void releaseClient() {
     }
   }
 
-  /**
-   * Updates DatanodeBlockId which based on blockId.
-   */
-  private void updateDatanodeBlockId(Pipeline pipeline) throws IOException {
+  private DatanodeBlockID buildDatanodeBlockId(Pipeline pipeline) throws IOException {
     DatanodeDetails closestNode = pipeline.getClosestNode();
     int replicaIdx = pipeline.getReplicaIndex(closestNode);
-    datanodeBlockID = blockID.getDatanodeBlockIDProtobufBuilder(replicaIdx).build();
-  }
-
-  private DatanodeBlockID buildDatanodeBlockId(Pipeline pipeline) throws IOException {
-    updateDatanodeBlockId(pipeline);
-    return datanodeBlockID;
+    return blockID.getDatanodeBlockIDProtobufBuilder(replicaIdx).build();
   }
 
   /**
    * Acquire new client if previous one was released.
-   *
-   * @return the held client after ensuring it is acquired
    */
-  private synchronized void acquireClient() throws IOException {
+  protected synchronized void acquireClient() throws IOException {
     if (xceiverClientFactory != null && xceiverClient == null) {
       Pipeline pipeline = pipelineSupplier.get();
       xceiverClient = xceiverClientFactory.acquireClientForReadData(pipeline);
-      updateDatanodeBlockId(pipeline);
+      datanodeBlockID = buildDatanodeBlockId(pipeline);
     }
   }
 
@@ -399,21 +391,7 @@ private synchronized void readChunkFromContainer(int len) throws IOException {
   }
 
   ChunkInfo getChunkInfo(long chunkRelativePosition, int toRead) {
-    final long adjustedOffset;
-    final long adjustedLen;
-    if (verifyChecksum) {
-      ChecksumBoundaries boundaries = computeChecksumBoundaries(chunkRelativePosition, toRead);
-      adjustedOffset = boundaries.offset;
-      adjustedLen = boundaries.length;
-    } else {
-      adjustedOffset = chunkRelativePosition;
-      adjustedLen = toRead;
-    }
-
-    return ChunkInfo.newBuilder(chunkInfo)
-        .setOffset(chunkInfo.getOffset() + adjustedOffset)
-        .setLen(adjustedLen)
-        .build();
+    return reader.getChunkInfo(chunkRelativePosition, toRead);
   }
 
   protected void readChunkDataIntoBuffers(ChunkInfo readChunkInfo)
@@ -433,79 +411,22 @@ protected void readChunkDataIntoBuffers(ChunkInfo readChunkInfo)
     allocated = 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 (!dst.hasRemaining()) {
-      return 0;
-    }
-    if (chunkRelativePosition < 0 || chunkRelativePosition >= length) {
-      return EOF;
-    }
-    final int toRead = (int) Math.min(dst.remaining(), length - chunkRelativePosition);
-
-    final ChunkInfo readChunkInfo = getChunkInfo(chunkRelativePosition, toRead);
-    final long adjustedOffset = readChunkInfo.getOffset() - chunkInfo.getOffset();
-
-    final long skip = chunkRelativePosition - adjustedOffset;
-    if (xceiverClientFactory == null) {
-      // for LocalChunkInputStream, which does not use a xceiver client, just read the chunk directly
-      return copyRange(readChunk(readChunkInfo), skip, toRead, dst);
-    }
+    return reader.read(chunkRelativePosition, dst);
+  }
 
+  /** Read a range with a client reference owned by this request. */
+  protected ByteBuffer[] readChunkPositioned(ChunkInfo range) throws IOException {
     final Pipeline pipeline = pipelineSupplier.get();
-    final ContainerProtos.DatanodeBlockID bid = buildDatanodeBlockId(pipeline);
+    final DatanodeBlockID bid = buildDatanodeBlockId(pipeline);
     final XceiverClientSpi client = xceiverClientFactory.acquireClientForReadData(pipeline);
     try {
-      final ByteBuffer[] readBuffers = readChunk(readChunkInfo, client, bid, validators, tokenSupplier.get());
-      return copyRange(readBuffers, skip, toRead, dst);
+      return readChunk(range, client, bid, validators, tokenSupplier.get());
     } finally {
       xceiverClientFactory.releaseClientForReadData(client, false);
     }
   }
 
-  /**
-   * 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;
-      }
-      if (remainingSkip > 0) {
-        if (remainingSkip >= buffer.remaining()) {
-          remainingSkip -= buffer.remaining();
-          continue;
-        } else {
-          buffer.position(Math.toIntExact(buffer.position() + remainingSkip));
-          remainingSkip = 0;
-        }
-      }
-      int n = Math.min(buffer.remaining(), toCopy - copied);
-      if (n <= 0) {
-        continue;
-      }
-      buffer.limit(buffer.position() + n);
-      dst.put(buffer);
-      copied += n;
-    }
-    return copied;
-  }
-
   /**
    * Send RPC call to get the chunk from the container using the sequential-read client held in
    * {@link #xceiverClient}. Called by the buffered sequential read path via
@@ -518,10 +439,7 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) throws IOException {
 
   /**
    * Send RPC call to get the chunk from the container using an explicitly provided client.
-   * Used by the positioned-read path so callers hold a local reference to the client
-   * rather than re-reading the shared {@link #xceiverClient} field after the lock is released.
-   * Subclasses that do not use an xceiver client (e.g. local short-circuit reads) should
-   * override this method to ignore {@code client} and read through their own mechanism.
+   * Shared by sequential reads and request-owned positioned reads.
    */
   private static ByteBuffer[] readChunk(
       ChunkInfo chunk, XceiverClientSpi client, DatanodeBlockID blockID,
@@ -591,50 +509,6 @@ private void validateChunk(
     }
   }
 
-  /**
-   * Return the offset and length of bytes that need to be read from the
-   * chunk file to cover the checksum boundaries covering the actual start and
-   * end of the chunk index to be read.
-   * For example, lets say the client is reading from index 120 to 450 in the
-   * chunk. And let's say checksum is stored for every 100 bytes in the chunk
-   * i.e. the first checksum is for bytes from index 0 to 99, the next for
-   * bytes from index 100 to 199 and so on. To verify bytes from 120 to 450,
-   * we would need to read from bytes 100 to 499 so that checksum
-   * verification can be done.
-   *
-   * @param startByteIndex the first byte index to be read by client
-   * @param dataLen number of bytes to be read from the chunk
-   * @return Adjusted (Chunk Offset, Chunk Length) which needs to be read
-   * from Container
-   */
-  private ChecksumBoundaries computeChecksumBoundaries(long startByteIndex, int dataLen) {
-
-    int bytesPerChecksum = chunkInfo.getChecksumData().getBytesPerChecksum();
-    // index of the last byte to be read from chunk, inclusively.
-    final long endByteIndex = startByteIndex + dataLen - 1;
-
-    long adjustedChunkOffset =  (startByteIndex / bytesPerChecksum)
-        * bytesPerChecksum; // inclusive
-    final long endIndex = ((endByteIndex / bytesPerChecksum) + 1)
-        * bytesPerChecksum; // exclusive
-    long adjustedChunkLen = Math.min(endIndex, length) - adjustedChunkOffset;
-    return new ChecksumBoundaries(adjustedChunkOffset, adjustedChunkLen);
-  }
-
-  /**
-   * Represents a byte range (offset and length) expanded to align with
-   * checksum chunk boundaries required for verification.
-   */
-  private static final class ChecksumBoundaries {
-    private final long offset;
-    private final long length;
-
-    private ChecksumBoundaries(long offset, long length) {
-      this.offset = offset;
-      this.length = length;
-    }
-  }
-
   /**
    * Adjust the buffers position to account for seeked position and/ or checksum
    * boundary reads.