From 1bd38aca0472cb1068ff9f0cd66348b99c0ea5d1 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Thu, 1 Oct 2026 13:29:08 +0800 Subject: [PATCH 1/2] HDDS-16325. Reuse chunk range readers for positioned reads --- .../hdds/scm/storage/BlockInputStream.java | 60 +++---- .../hdds/scm/storage/ChunkInputStream.java | 152 ++---------------- .../hadoop/hdds/scm/storage/ChunkReader.java | 129 +++++++++++++++ .../scm/storage/LocalChunkInputStream.java | 19 ++- .../scm/storage/DummyChunkInputStream.java | 5 + .../scm/storage/TestChunkInputStream.java | 68 ++++++++ 6 files changed, 250 insertions(+), 183 deletions(-) create mode 100644 hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkReader.java diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java index c3b1f3ed73f1..7bfc046732c4 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java +++ b/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,7 +485,7 @@ public int readPositioned(long blockRelativePosition, ByteBuffer dst) throws IOE initialize(); } final long[] offsets; - final BlockData currentBlockData; + final List streams; final long blockLength; synchronized (this) { checkOpen(); @@ -494,22 +493,21 @@ public int readPositioned(long blockRelativePosition, ByteBuffer dst) throws IOE return EOF; } - offsets = Arrays.copyOf(chunkOffsets, chunkOffsets.length); - currentBlockData = blockData; + offsets = chunkOffsets; + streams = chunkStreams; blockLength = length; } - final List 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; } } diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java index 67f769f56e47..26d03f25877b 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java +++ b/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 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,18 +295,10 @@ 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(); } /** @@ -312,11 +306,11 @@ private DatanodeBlockID buildDatanodeBlockId(Pipeline pipeline) throws IOExcepti * * @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 +393,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 +413,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 (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(); - - 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 +441,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 +511,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. diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkReader.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkReader.java new file mode 100644 index 000000000000..14c5fb6ce74c --- /dev/null +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkReader.java @@ -0,0 +1,129 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.hdds.scm.storage; + +import java.io.IOException; +import java.nio.ByteBuffer; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; +import org.apache.ratis.util.function.CheckedFunction; + +/** + * Reads checksum-aligned ranges without retaining a cursor or data buffers. + */ +final class ChunkReader { + private final ChunkInfo chunkInfo; + private final long length; + private final boolean verifyChecksum; + private final CheckedFunction readChunk; + + ChunkReader(ChunkInfo chunkInfo, boolean verifyChecksum, + CheckedFunction readChunk) { + this.chunkInfo = chunkInfo; + this.length = chunkInfo.getLen(); + this.verifyChecksum = verifyChecksum; + this.readChunk = readChunk; + } + + int read(long position, ByteBuffer destination) throws IOException { + if (!destination.hasRemaining()) { + return 0; + } + if (position < 0 || position >= length) { + return -1; + } + final int toRead = (int) Math.min(destination.remaining(), length - position); + final ChunkInfo range = getChunkInfo(position, toRead); + final long skip = position - (range.getOffset() - chunkInfo.getOffset()); + return copyRange(readChunk.apply(range), skip, toRead, destination); + } + + 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(); + } + + 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; + } + } + + private static int copyRange(ByteBuffer[] src, long skip, int toCopy, ByteBuffer dst) { + long remainingSkip = skip; + int copied = 0; + for (ByteBuffer source : src) { + ByteBuffer buffer = source.duplicate(); + 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; + } + +} diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java index a6e429fd6e3d..8fb8487b31b0 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java @@ -19,7 +19,6 @@ import com.google.common.annotations.VisibleForTesting; import java.io.EOFException; -import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; import java.nio.ByteBuffer; @@ -62,12 +61,10 @@ public class LocalChunkInputStream extends ChunkInputStream @SuppressWarnings("checkstyle:parameternumber") LocalChunkInputStream(ChunkInfo chunkInfo, BlockID blockId, XceiverClientFactory xceiverClientFactory, Supplier pipelineSupplier, boolean verifyChecksum, Supplier> tokenSupplier, - XceiverClientShortCircuit xceiverClientShortCircuit, FileInputStream blockInputStream) { - // Pass null for xceiverClientFactory such readPositioned() is taken, both routing reads - // and checksum verification are done locally. - super(chunkInfo, blockId, null, pipelineSupplier, verifyChecksum, tokenSupplier); + XceiverClientShortCircuit xceiverClientShortCircuit, FileChannel dataIn) { + super(chunkInfo, blockId, xceiverClientFactory, pipelineSupplier, verifyChecksum, tokenSupplier); this.chunkInfo = chunkInfo; - this.dataIn = blockInputStream.getChannel(); + this.dataIn = dataIn; this.validator = this::validateChunk; this.verifyChecksum = verifyChecksum; if (LOG.isDebugEnabled()) { @@ -75,6 +72,16 @@ public class LocalChunkInputStream extends ChunkInputStream } } + @Override + protected void acquireClient() { + // The block stream owns the local file descriptor. + } + + @Override + protected ByteBuffer[] readChunkPositioned(ChunkInfo range) throws IOException { + return readChunk(range); + } + /** * Get the chunk from the local block replica. */ diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java index 97ec08f4d12f..b9d2adc528dd 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java @@ -78,6 +78,11 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) { .toArray(new ByteBuffer[0]); } + @Override + protected ByteBuffer[] readChunkPositioned(ChunkInfo range) { + return readChunk(range); + } + @Override protected void releaseClient() { // no-op diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java index 96d911319f05..8abd9eb8b13c 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java @@ -25,15 +25,26 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.EOFException; +import java.io.IOException; import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Arrays; import java.util.List; import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import org.apache.hadoop.hdds.client.BlockID; @@ -51,6 +62,7 @@ import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import org.mockito.ArgumentCaptor; /** @@ -69,6 +81,22 @@ public class TestChunkInputStream { private ChunkInfo chunkInfo; private byte[] chunkData; + @Test + void concurrentLocalRangeReadsPreserveFileAndStreamPositions(@TempDir Path directory) throws Exception { + byte[] data = generateRandomData(PositionedReadTestHelper.SOURCE_SIZE); + Path path = Files.write(directory.resolve("block"), data); + ChunkInfo info = chunkInfo.toBuilder().setLen(data.length) + .setChecksumData(new Checksum(ChecksumType.CRC32, 4096).computeChecksum(data).getProtoBufMessage()).build(); + try (FileChannel file = FileChannel.open(path); + LocalChunkInputStream stream = new LocalChunkInputStream(info, blockID, null, () -> null, true, + () -> null, null, file)) { + stream.seek(123); + PositionedReadTestHelper.runConcurrentPositionedReads(data, stream::readPositioned); + assertEquals(123, stream.getPos()); + assertEquals(0, file.position()); + } + } + @BeforeEach public void setup() throws Exception { Checksum checksum = new Checksum(ChecksumType.CRC32, BYTES_PER_CHECKSUM); @@ -308,6 +336,46 @@ public void testUnbuffered() throws Exception { matchWithInputData(b2, 20, 20); } + @Test + void rangeReadsOverlapWithoutChangingSourceBufferPositions() throws Exception { + ByteBuffer source = ByteBuffer.wrap(chunkData, 0, BYTES_PER_CHECKSUM).slice().asReadOnlyBuffer(); + CountDownLatch entered = new CountDownLatch(2); + ChunkReader reader = new ChunkReader(chunkInfo, true, range -> { + entered.countDown(); + try { + if (!entered.await(5, TimeUnit.SECONDS)) { + throw new IOException("Range reads were serialized"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + return new ByteBuffer[] {source}; + }); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future first = executor.submit(() -> { + ByteBuffer destination = ByteBuffer.allocate(7); + assertEquals(7, reader.read(3, destination)); + assertArrayEquals(Arrays.copyOfRange(chunkData, 3, 10), destination.array()); + return null; + }); + Future second = executor.submit(() -> { + ByteBuffer destination = ByteBuffer.allocate(7); + assertEquals(7, reader.read(12, destination)); + assertArrayEquals(Arrays.copyOfRange(chunkData, 12, 19), destination.array()); + return null; + }); + first.get(10, TimeUnit.SECONDS); + second.get(10, TimeUnit.SECONDS); + assertEquals(0, source.position()); + assertEquals(BYTES_PER_CHECKSUM, source.limit()); + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + @Test public void connectsToNewPipeline() throws Exception { // GIVEN From 7fe1f2167876affd3da3e032a672a410cf168747 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Tue, 6 Oct 2026 13:50:22 +0800 Subject: [PATCH 2/2] HDDS-16325. Remove invalid return tag from acquireClient Javadoc --- .../org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java index 26d03f25877b..d538e88ec3de 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java @@ -303,8 +303,6 @@ private DatanodeBlockID buildDatanodeBlockId(Pipeline pipeline) throws IOExcepti /** * Acquire new client if previous one was released. - * - * @return the held client after ensuring it is acquired */ protected synchronized void acquireClient() throws IOException { if (xceiverClientFactory != null && xceiverClient == null) {