szetszwo commented on code in PR #11102:
URL: https://github.com/apache/ozone/pull/11102#discussion_r3882838258
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -486,6 +487,106 @@ 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 = chunkOffsets;
+ final BlockData currentBlockData = blockData;
+ final long blockLength = length;
Review Comment:
They need to be synchronized.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -486,6 +487,106 @@ 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 = chunkOffsets;
+ final BlockData currentBlockData = blockData;
+ final long 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)
+ throws IOException {
+ final int startPosition = dst.position();
+ int preadRetries = 0;
+ while (true) {
+ final ChunkInputStream chunkStream = createChunkInputStream(chunkInfo);
Review Comment:
createChunkInputStream uses xceiverClientFactory, xceiverClientGrpc and
xceiverClientShortCircuit, which need to be synchronized.
--
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]
