rich7420 commented on code in PR #11102:
URL: https://github.com/apache/ozone/pull/11102#discussion_r3864600966
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java:
##########
@@ -429,6 +429,108 @@ 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}, which reads from a shared {@link
+ * java.nio.channels.FileChannel} and therefore must serialize.
+ */
+ 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) {
Review Comment:
Could we synchronize local positioned reads on the shared block FileChannel,
or use positional FileChannel reads? Each LocalChunkInputStream locks its own
instance, while all chunks in the block receive the same blockFileInputStream,
so preads to different chunks can still interleave position(...).read(...) and
use the wrong offset.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -486,6 +487,57 @@ 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 chunk streams' buffered state, so it is safe
+ * for concurrent callers. Metadata ({@code chunkOffsets}, {@code
+ * chunkStreams}, {@code length}) is published once by {@link #initialize()};
+ * {@link #initialized} is {@code volatile} so callers observe a consistent
+ * snapshot after initialization completes.
+ *
+ * @return bytes copied into {@code dst}, or {@link #EOF} at EOF
+ */
+ int readPositioned(long blockRelativePosition, ByteBuffer dst)
+ throws IOException {
+ if (!initialized) {
+ initialize();
+ }
+ final List<ChunkInputStream> streams = chunkStreams;
+ final long[] offsets = chunkOffsets;
+ final long blockLength = length;
+ if (streams == null || streams.isEmpty()
+ || blockRelativePosition < 0 || blockRelativePosition >= blockLength) {
+ return EOF;
+ }
+
+ int total = 0;
+ long pos = blockRelativePosition;
+ while (dst.hasRemaining() && pos < blockLength) {
+ int idx = chunkIndexForPosition(pos, offsets);
+ ChunkInputStream chunk = streams.get(idx);
+ long chunkPos = pos - offsets[idx];
+ int n = chunk.readPositioned(chunkPos, dst);
Review Comment:
Could we preserve BlockInputStream’s outer retry and refresh handling here?
readChunk still tries the DNs in the current pipeline, but this native path
cannot refresh an expired block token or fetch an updated pipeline after those
attempts fail.
--
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]
