szetszwo commented on code in PR #11102:
URL: https://github.com/apache/ozone/pull/11102#discussion_r3981696662
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java:
##########
@@ -430,14 +433,130 @@ protected void readChunkDataIntoBuffers(ChunkInfo
readChunkInfo)
}
/**
- * Send RPC call to get the chunk from the container.
+ * 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.
*/
- @VisibleForTesting
- protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
+ 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;
+ }
+
+ 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();
Review Comment:
Let's add a getChunkInfo(..) method and use it here and also
readChunkFromContainer(..):
```diff
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
@@ -381,29 +381,10 @@ private synchronized void readChunkFromContainer(int
len) throws IOException {
// successful read in adjustBufferPosition()
storePosition();
- long adjustedBuffersOffset, adjustedBuffersLen;
- if (verifyChecksum) {
- // Adjust the chunk offset and length to include required checksum
- // boundaries
- Pair<Long, Long> adjustedOffsetAndLength =
- computeChecksumBoundaries(startByteIndex, len);
- adjustedBuffersOffset = adjustedOffsetAndLength.getLeft();
- adjustedBuffersLen = adjustedOffsetAndLength.getRight();
- } else {
- // Read from the startByteIndex
- adjustedBuffersOffset = startByteIndex;
- adjustedBuffersLen = len;
- }
-
- // Adjust the chunkInfo so that only the required bytes are read from
- // the chunk.
- final ChunkInfo adjustedChunkInfo = ChunkInfo.newBuilder(chunkInfo)
- .setOffset(chunkInfo.getOffset() + adjustedBuffersOffset)
- .setLen(adjustedBuffersLen)
- .build();
+ final ChunkInfo adjustedChunkInfo = getChunkInfo(startByteIndex, len);
readChunkDataIntoBuffers(adjustedChunkInfo);
- bufferOffsetWrtChunkData = adjustedBuffersOffset;
+ bufferOffsetWrtChunkData = adjustedChunkInfo.getOffset() -
chunkInfo.getOffset();
// If the stream was seeked to position before, then the buffer
// position should be adjusted as the reads happen at checksum
boundaries.
```
```diff
@@ -415,6 +396,24 @@ private synchronized void readChunkFromContainer(int
len) throws IOException {
adjustBufferPosition(startByteIndex - bufferOffsetWrtChunkData);
}
+ ChunkInfo getChunkInfo(long chunkRelativePosition, int toRead) {
+ 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;
+ }
+
+ return ChunkInfo.newBuilder(chunkInfo)
+ .setOffset(chunkInfo.getOffset() + adjustedOffset)
+ .setLen(adjustedLen)
+ .build();
+ }
+
protected void readChunkDataIntoBuffers(ChunkInfo readChunkInfo)
throws IOException {
```
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java:
##########
@@ -219,6 +234,60 @@ int readImpl(InputStream inputStream) throws IOException {
return true;
}
+ /**
+ * Stateless positioned read for the replicated (Ratis) path. Routes the read
+ * across the part {@link BlockInputStream}s using the immutable
+ * {@link #partOffsets} without seeking or mutating the shared cursor, so
+ * concurrent positioned reads run independently. Returns {@code false} (so
+ * the caller can fall back) when any part is not a {@link BlockInputStream},
+ * e.g. erasure coded parts.
+ */
+ private boolean readFullyStateless(long position, ByteBuffer buffer)
+ throws IOException {
+ if (!buffer.hasRemaining()) {
+ return true;
+ }
+ if (!statelessPositionedReadSupported) {
+ return false;
+ }
+
+ 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);
+ }
+ int idx = partIndexForPosition(pos);
+ BlockInputStream part = (BlockInputStream) partStreams.get(idx);
+ long partPos = pos - partOffsets[idx];
+ int n = part.readPositioned(partPos, buffer);
+ if (n <= 0) {
+ if (bytesRead > 0) {
+ return true;
+ }
+ throw new EOFException("EOF encountered at pos: " + pos +
+ " for key: " + key);
+ }
+ bytesRead += n;
+ pos += n;
+ }
+ return true;
+ }
+
+ private int partIndexForPosition(long pos) {
+ int idx = Arrays.binarySearch(partOffsets, pos);
+ if (idx < 0) {
+ // binarySearch returns -insertionPoint - 1; the containing part is
+ // insertionPoint - 1.
+ idx = -idx - 2;
+ }
+ return idx;
+ }
Review Comment:
Let's make it static and reuse it in BlockInputStream:
```java
static int binarySearchOffsetIndex(long[] offsets, long pos) {
final int idx = Arrays.binarySearch(offsets, pos);
if (idx > 0) {
return idx;
}
// binarySearch returns n = -insertionPoint - 1;
// insertionPoint is -n - 1
// the containing index is insertionPoint - 1.
return -idx - 2;
}
```
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java:
##########
@@ -430,14 +433,130 @@ protected void readChunkDataIntoBuffers(ChunkInfo
readChunkInfo)
}
/**
- * Send RPC call to get the chunk from the container.
+ * 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.
*/
- @VisibleForTesting
- protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
+ boolean supportsConcurrentPositionedRead() {
+ return true;
+ }
Review Comment:
supportsConcurrentPositionedRead() always returns true. Let's remove it:
```diff
@@ -432,18 +431,6 @@ 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
```
```diff
@@ -456,19 +443,7 @@ boolean supportsConcurrentPositionedRead() {
* @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 {
+ synchronized int readPositioned(long chunkRelativePosition, ByteBuffer
dst) throws IOException {
if (chunkRelativePosition < 0 || chunkRelativePosition >= length) {
return EOF;
}
```
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -486,6 +487,116 @@ 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 {
Review Comment:
Check blockRelativePosition first. BTW, please don't use short lines:
```java
int readPositioned(long blockRelativePosition, ByteBuffer dst) throws
IOException {
if (blockRelativePosition < 0) {
return EOF;
}
```
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java:
##########
@@ -306,13 +306,16 @@ private void updateDatanodeBlockId(Pipeline pipeline)
throws IOException {
/**
* Acquire new client if previous one was released.
+ *
+ * @return the held client after ensuring it is acquired
*/
- protected synchronized void acquireClient() throws IOException {
+ protected synchronized XceiverClientSpi acquireClient() throws IOException {
Review Comment:
Change it to private and remove it from LocalChunkInputStream and
DummyChunkInputStream.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java:
##########
@@ -430,14 +433,130 @@ protected void readChunkDataIntoBuffers(ChunkInfo
readChunkInfo)
}
/**
- * Send RPC call to get the chunk from the container.
+ * 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.
*/
- @VisibleForTesting
- protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
+ 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;
+ }
+
+ 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();
+
+ // Capture the client reference under the acquireClient() lock so that a
concurrent
+ // releaseClient() (e.g. from the sequential read's handleReadError or
unbuffer) cannot
+ // null the xceiverClient field between acquisition and use.
+ final XceiverClientSpi client = acquireClient();
+ final ByteBuffer[] readBuffers = readChunk(client, 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;
+ }
+ }
Review Comment:
- duplicate() is not needed
- We may simplify it by first checking remainingSkip >= buffer.remaining().
```java
if (remainingSkip > 0) {
if (remainingSkip >= buffer.remaining()) {
remainingSkip -= buffer.remaining();
continue;
} else {
buffer.position(Math.toIntExact(buffer.position() +
remainingSkip));
remainingSkip = 0;
}
}
```
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -486,6 +487,116 @@ 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;
+ final BlockData currentBlockData;
+ final long blockLength;
+ synchronized (this) {
+ checkOpen();
+ offsets = chunkOffsets;
+ currentBlockData = blockData;
+ blockLength = length;
+ }
+ if (offsets == null || currentBlockData == null
+ || blockRelativePosition < 0 || blockRelativePosition >= blockLength) {
Review Comment:
- chunkOffsets is an array. We need to copy it:
- Let's move the if inside since we have to make use chunkOffsets is not
null before copying.
```java
synchronized (this) {
checkOpen();
if (chunkOffsets == null || blockData == null || blockRelativePosition
>= length) {
return EOF;
}
offsets = Arrays.copyOf(chunkOffsets, chunkOffsets.length);
currentBlockData = blockData;
blockLength = length;
}
```
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -486,6 +487,116 @@ 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
Review Comment:
This part of the code is inserted between seek(..) and seek(..)'s javadoc.
Please keep seek(..)'s javadoc with seek(..).
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -486,6 +487,116 @@ 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;
+ final BlockData currentBlockData;
+ final long blockLength;
+ synchronized (this) {
+ checkOpen();
+ offsets = chunkOffsets;
+ currentBlockData = blockData;
+ 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)
Review Comment:
Let's rename numBytesToRead to readLength (or toRead) since numBytesToRead
and numBytesRead look similar.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java:
##########
@@ -430,14 +433,130 @@ protected void readChunkDataIntoBuffers(ChunkInfo
readChunkInfo)
}
/**
- * Send RPC call to get the chunk from the container.
+ * 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.
*/
- @VisibleForTesting
- protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
+ 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;
+ }
+
+ 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();
+
+ // Capture the client reference under the acquireClient() lock so that a
concurrent
+ // releaseClient() (e.g. from the sequential read's handleReadError or
unbuffer) cannot
+ // null the xceiverClient field between acquisition and use.
+ final XceiverClientSpi client = acquireClient();
+ final ByteBuffer[] readBuffers = readChunk(client, 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 using the
sequential-read client held in
+ * {@link #xceiverClient}. Called by the buffered sequential read path via
+ * {@link #readChunkDataIntoBuffers}.
+ */
+ @VisibleForTesting
+ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) throws IOException
{
+ return readChunk(xceiverClient, readChunkInfo);
+ }
+ /**
+ * 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.
+ */
+ protected ByteBuffer[] readChunk(XceiverClientSpi client, ChunkInfo
readChunkInfo)
Review Comment:
- Make it private.
- Reorder the parameters so both readChunk methods start with readChunkInfo.
- Avoid short lines.
```java
/**
* 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.
*/
private ByteBuffer[] readChunk(ChunkInfo readChunkInfo, XceiverClientSpi
client) throws IOException {
```
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java:
##########
@@ -82,12 +88,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;
+ }
+ }
+ }
Review Comment:
- We should keep looping for the same buffer until it is filled up.
- We should throw EOFException when a buffer cannot be filled up.
```java
private void readAtOffset(ByteBuffer[] buffers, long fileOffset) throws
IOException {
long pos = fileOffset;
for (ByteBuffer buffer : buffers) {
final int remaining = buffer.remaining();
final long read = readAtOffset(buffer, pos);
if (buffer.hasRemaining()) {
throw new EOFException("Read only " + read + "bytes but expected to
read " + remaining
+ " bytes at offset " + pos + " for chunk " +
chunkInfo.getChunkName());
}
pos += read;
}
}
```
```java
private long readAtOffset(ByteBuffer buffer, long fileOffset) throws
IOException {
long pos = fileOffset;
while (buffer.hasRemaining()) {
final int n = dataIn.read(buffer, pos);
if (n < 0) {
break;
}
pos += n;
}
return pos - fileOffset;
}
```
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java:
##########
@@ -330,8 +330,9 @@ protected synchronized void acquireClient() throws
IOException {
throw new IOException("Unexpected client class: " +
client.getClass().getName() + ", " + pipeline);
}
- xceiverClient = (XceiverClientGrpc) client;
+ xceiverClient = (XceiverClientGrpc) client;
}
+ return xceiverClient;
Review Comment:
All the changes in this file are not needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
