taklwu commented on code in PR #11102:
URL: https://github.com/apache/ozone/pull/11102#discussion_r3986162301
##########
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:
the tests with `DummyChunkInputStream` and the actual use case of
`LocalChunkInputStream` may fail with NPE for `datanodeBlockID` if we change
this back to private
```
readPositioned
→ acquireClient() // no-op for Dummy/Local (xceiverClientFactory
== null)
// xceiverClient stays null, datanodeBlockID
stays null
→ readChunk(readChunkInfo, client) // PRIVATE method, statically bound
↓ (no polymorphism: private = no vtable entry)
ChunkInputStream.readChunk(ChunkInfo, XceiverClientSpi) ← always this,
even on DummyChunkInputStream
↓
ContainerProtocolCalls.readChunk(client, readChunkInfo, datanodeBlockID,
...)
```
we may need to keep this as protected and override in
`DummyChunkInputStream` and `LocalChunkInputStream` that call the
`readChunk(ChunkInfo)` without the client explicitly.
--
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]
