This is an automated email from the ASF dual-hosted git repository.
smengcl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 0d57c08f741 HDDS-16620. Expose BlockReadCursor as an iterator of
streaming ReadBlock ranges (#11442)
0d57c08f741 is described below
commit 0d57c08f74162fe6f92cb1523816dbead0fb7d6d
Author: Peter Lee <[email protected]>
AuthorDate: Thu Oct 8 14:23:45 2026 +0800
HDDS-16620. Expose BlockReadCursor as an iterator of streaming ReadBlock
ranges (#11442)
---
.../ozone/container/keyvalue/BlockReadCursor.java | 75 ++++++++++++-------
.../ozone/container/keyvalue/KeyValueHandler.java | 20 ++---
.../container/keyvalue/TestBlockReadCursor.java | 86 ++++++++++++----------
3 files changed, 107 insertions(+), 74 deletions(-)
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/BlockReadCursor.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/BlockReadCursor.java
index e877039829a..3e4567cabf8 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/BlockReadCursor.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/BlockReadCursor.java
@@ -19,19 +19,21 @@
import java.io.IOException;
import java.util.Arrays;
+import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
-/** Tracks chunk-relative checksum boundaries for a streaming block read. */
-class BlockReadCursor {
+/** Splits a streaming block read into ranges aligned to chunk-relative
checksum boundaries. */
+class BlockReadCursor implements Iterator<BlockReadCursor.ReadRange> {
private static final int STREAMING_BYTES_PER_CHUNK = 1024 * 64;
private final List<ChunkInfo> chunks;
// [c1 offset, c2 offset, ..., cn offset, cn offset + cn len]
private final long[] chunkOffsets;
+ private final int[] intervals;
private final int responseDataSize;
- private final long start;
private final long end;
private long offset;
private int chunkIndex;
@@ -39,27 +41,28 @@ class BlockReadCursor {
BlockReadCursor(long requestedOffset, long length, int responseSize,
List<ChunkInfo> chunks) throws IOException {
this.chunks = chunks;
chunkOffsets = new long[chunks.size() + 1];
+ intervals = new int[chunks.size()];
int bufferSize = responseSize;
for (int i = 0; i < chunks.size(); i++) {
ChunkInfo chunk = chunks.get(i);
chunkOffsets[i] = chunk.getOffset();
+ intervals[i] = interval(chunk);
// One checksum interval (or the short chunk containing it) must always
fit in the buffer.
- bufferSize = Math.max(bufferSize, (int) Math.min(chunk.getLen(),
interval(chunk)));
+ bufferSize = Math.max(bufferSize, (int) Math.min(chunk.getLen(),
intervals[i]));
}
ChunkInfo lastChunk = chunks.get(chunks.size() - 1);
long blockEnd = lastChunk.getOffset() + lastChunk.getLen();
chunkOffsets[chunks.size()] = blockEnd;
this.responseDataSize = bufferSize;
chunkIndex = findChunk(requestedOffset);
- start = length == 0 ? requestedOffset : alignDown(requestedOffset,
chunkIndex);
- offset = start;
+ offset = length == 0 ? requestedOffset : alignDown(requestedOffset,
chunkIndex);
if (length == 0) {
- end = start;
+ end = offset;
} else {
long requestedEnd = requestedOffset + Math.min(length, blockEnd -
requestedOffset);
int last = findChunk(requestedEnd - 1);
long floor = alignDown(requestedEnd - 1, last);
- end = floor + Math.min(interval(chunks.get(last)), chunkOffsets[last +
1] - floor);
+ end = floor + Math.min(intervals[last], chunkOffsets[last + 1] - floor);
}
}
@@ -75,8 +78,8 @@ private static int interval(ChunkInfo chunk) throws
IOException {
return size;
}
- private long alignDown(long position, int index) throws IOException {
- return position - (position - chunkOffsets[index]) %
interval(chunks.get(index));
+ private long alignDown(long position, int index) {
+ return position - (position - chunkOffsets[index]) % intervals[index];
}
private int findChunk(long position) {
@@ -88,11 +91,8 @@ private int findChunk(long position) {
return insertionPoint - 1;
}
- long offset() {
- return offset;
- }
-
- boolean hasRemaining() {
+ @Override
+ public boolean hasNext() {
return offset < end;
}
@@ -100,26 +100,47 @@ int responseDataSize() {
return responseDataSize;
}
- int nextReadLength() throws IOException {
+ /** Hands out the next range and advances past it, whether or not the caller
reads it. */
+ @Override
+ public ReadRange next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException("No range left at offset " + offset + "
of " + end);
+ }
long limit = offset + Math.min(responseDataSize, end - offset);
if (limit < end) {
limit = alignDown(limit, findChunk(limit));
}
- return Math.toIntExact(limit - offset);
+ ReadRange range = new ReadRange(offset, Math.toIntExact(limit - offset),
+ chunks.subList(chunkIndex, findChunk(limit - 1) + 1));
+ offset = limit;
+ if (hasNext()) {
+ chunkIndex = findChunk(offset);
+ }
+ return range;
}
- List<ChunkInfo> chunksForRead(int length) {
- return chunks.subList(chunkIndex, findChunk(offset + length - 1) + 1);
- }
+ /** A range of the block sent in one response, with the chunks it overlaps.
*/
+ static final class ReadRange {
+ private final long offset;
+ private final int length;
+ private final List<ChunkInfo> chunks;
- void advance(int bytesRead) {
- offset += bytesRead;
- if (hasRemaining()) {
- chunkIndex = findChunk(offset);
+ private ReadRange(long offset, int length, List<ChunkInfo> chunks) {
+ this.offset = offset;
+ this.length = length;
+ this.chunks = chunks;
+ }
+
+ long offset() {
+ return offset;
}
- }
- long bytesRead() {
- return offset - start;
+ int length() {
+ return length;
+ }
+
+ List<ChunkInfo> chunks() {
+ return chunks;
+ }
}
}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
index b4342fcdc91..0ec6965f358 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
@@ -2348,23 +2348,23 @@ private long readBlockImpl(ContainerCommandRequestProto
request, RandomAccessFil
final BlockReadCursor cursor = new BlockReadCursor(readBlock.getOffset(),
readBlock.getLength(),
responseDataSize, blockData.getChunks());
final ByteBuffer buffer = ByteBuffer.allocate(cursor.responseDataSize());
- blockFile.position(cursor.offset());
- while (cursor.hasRemaining()) {
- buffer.clear().limit(cursor.nextReadLength());
+ long bytesRead = 0;
+ while (cursor.hasNext()) {
+ final BlockReadCursor.ReadRange range = cursor.next();
+ blockFile.position(range.offset());
+ buffer.clear().limit(range.length());
blockFile.read(buffer);
if (buffer.hasRemaining()) {
- throw new EOFException("Unexpected end of block " + blockID + " at " +
cursor.offset());
+ throw new EOFException("Unexpected end of block " + blockID + " at " +
range.offset());
}
+ bytesRead += range.length();
buffer.flip();
- final int readLength = buffer.remaining();
- final List<ContainerProtos.ChunkInfo> chunks =
cursor.chunksForRead(readLength);
if (validateChunkChecksumData) {
- Checksum.validateChecksums(buffer, cursor.offset(), 0, chunks);
+ Checksum.validateChecksums(buffer, range.offset(), 0, range.chunks());
}
- streamObserver.onNext(getReadBlockResponse(request, chunks, buffer,
cursor.offset()));
- cursor.advance(readLength);
+ streamObserver.onNext(getReadBlockResponse(request, range.chunks(),
buffer, range.offset()));
}
- return cursor.bytesRead();
+ return bytesRead;
}
/**
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestBlockReadCursor.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestBlockReadCursor.java
index 1dda101b8c7..f7360f1a86c 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestBlockReadCursor.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestBlockReadCursor.java
@@ -19,11 +19,14 @@
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 java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.NoSuchElementException;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
import org.junit.jupiter.api.Test;
@@ -36,36 +39,51 @@ private static ChunkInfo chunk(long offset, long length,
int interval) {
.setType(ContainerProtos.ChecksumType.CRC32).setBytesPerChecksum(interval)).build();
}
+ private static void assertNextRange(BlockReadCursor cursor, long offset, int
length, List<ChunkInfo> chunks) {
+ assertTrue(cursor.hasNext());
+ BlockReadCursor.ReadRange range = cursor.next();
+ assertEquals(offset, range.offset());
+ assertEquals(length, range.length());
+ assertEquals(chunks, range.chunks());
+ }
+
@Test
void testShiftedRangesAndLookahead() throws Exception {
List<ChunkInfo> chunks = Arrays.asList(chunk(0, 3, 4), chunk(3, 9, 4),
chunk(12, 7, 3));
BlockReadCursor cursor = new BlockReadCursor(4, 14, 6, chunks);
- assertEquals(3, cursor.offset());
- assertEquals(4, cursor.nextReadLength());
- assertEquals(4, cursor.nextReadLength());
- assertEquals(chunks.subList(1, 2), cursor.chunksForRead(4));
- cursor.advance(4);
- assertEquals(7, cursor.offset());
- assertEquals(5, cursor.nextReadLength());
- cursor.advance(5);
- assertEquals(12, cursor.offset());
- assertEquals(6, cursor.nextReadLength());
- cursor.advance(6);
- assertFalse(cursor.hasRemaining());
- assertEquals(15, cursor.bytesRead());
+ assertTrue(cursor.hasNext());
+ assertNextRange(cursor, 3, 4, chunks.subList(1, 2));
+ assertNextRange(cursor, 7, 5, chunks.subList(1, 2));
+ assertNextRange(cursor, 12, 6, chunks.subList(2, 3));
+ assertFalse(cursor.hasNext());
}
@Test
void testSpanningChunksAndShortFinalChecksum() throws Exception {
List<ChunkInfo> chunks = Arrays.asList(chunk(0, 3, 4), chunk(3, 9, 4));
BlockReadCursor cursor = new BlockReadCursor(0, Long.MAX_VALUE, 8, chunks);
- assertEquals(7, cursor.nextReadLength());
- assertEquals(chunks, cursor.chunksForRead(7));
- cursor.advance(7);
- assertEquals(5, cursor.nextReadLength());
- cursor.advance(5);
- assertFalse(cursor.hasRemaining());
- assertEquals(12, cursor.bytesRead());
+ assertNextRange(cursor, 0, 7, chunks);
+ assertNextRange(cursor, 7, 5, chunks.subList(1, 2));
+ assertFalse(cursor.hasNext());
+ }
+
+ @Test
+ void testIteratesWholeRangeOnce() throws Exception {
+ List<ChunkInfo> chunks = Arrays.asList(chunk(0, 3, 4), chunk(3, 9, 4),
chunk(12, 7, 3));
+ BlockReadCursor cursor = new BlockReadCursor(0, Long.MAX_VALUE, 5, chunks);
+ List<Integer> lengths = new ArrayList<>();
+ long offset = 0;
+ while (cursor.hasNext()) {
+ BlockReadCursor.ReadRange range = cursor.next();
+ assertEquals(offset, range.offset());
+ lengths.add(range.length());
+ offset += range.length();
+ }
+ assertEquals(Arrays.asList(3, 4, 5, 3, 4), lengths);
+ assertFalse(cursor.hasNext());
+ assertThrows(NoSuchElementException.class, cursor::next);
+ assertFalse(cursor.hasNext());
+ assertThrows(NoSuchElementException.class, new BlockReadCursor(3, 0, 1,
chunks)::next);
}
@Test
@@ -75,12 +93,8 @@ void testReadsAtChunkAndChecksumBoundaries() throws
Exception {
{12, 15, 2}, {15, 18, 2}, {18, 19, 2}}) {
for (int requestedOffset : new int[] {range[0], range[1] - 1}) {
BlockReadCursor cursor = new BlockReadCursor(requestedOffset, 1, 8,
chunks);
- int length = range[1] - range[0];
- assertEquals(range[0], cursor.offset());
- assertEquals(length, cursor.nextReadLength());
- assertEquals(Collections.singletonList(chunks.get(range[2])),
cursor.chunksForRead(length));
- cursor.advance(length);
- assertFalse(cursor.hasRemaining());
+ assertNextRange(cursor, range[0], range[1] - range[0],
Collections.singletonList(chunks.get(range[2])));
+ assertFalse(cursor.hasNext());
}
}
}
@@ -90,29 +104,27 @@ void testMinimumBufferAndExactChunkEnd() throws Exception {
List<ChunkInfo> chunks = Arrays.asList(chunk(0, 3, 4), chunk(3, 9, 4));
BlockReadCursor cursor = new BlockReadCursor(3, 9, 1, chunks);
for (int expected : new int[] {4, 4, 1}) {
- assertTrue(cursor.hasRemaining());
- assertEquals(expected, cursor.nextReadLength());
- cursor.advance(expected);
+ assertTrue(cursor.hasNext());
+ assertEquals(expected, cursor.next().length());
}
- assertFalse(cursor.hasRemaining());
- assertFalse(new BlockReadCursor(3, 0, 1, chunks).hasRemaining());
+ assertFalse(cursor.hasNext());
+ assertFalse(new BlockReadCursor(3, 0, 1, chunks).hasNext());
BlockReadCursor shortChunk = new BlockReadCursor(0, 3, 1,
Collections.singletonList(chunk(0, 3, Integer.MAX_VALUE)));
assertEquals(3, shortChunk.responseDataSize());
- assertEquals(3, shortChunk.nextReadLength());
+ assertEquals(3, shortChunk.next().length());
}
@Test
void testRangeNearLongLimit() throws Exception {
List<ChunkInfo> chunks = Arrays.asList(chunk(0, Long.MAX_VALUE - 9, 4),
chunk(Long.MAX_VALUE - 9, 9, 4));
BlockReadCursor cursor = new BlockReadCursor(Long.MAX_VALUE - 8,
Long.MAX_VALUE, 4, chunks);
- assertEquals(Long.MAX_VALUE - 9, cursor.offset());
+ long offset = Long.MAX_VALUE - 9;
for (int length : new int[] {4, 4, 1}) {
- assertEquals(length, cursor.nextReadLength());
- cursor.advance(length);
+ assertNextRange(cursor, offset, length, chunks.subList(1, 2));
+ offset += length;
}
- assertEquals(Long.MAX_VALUE, cursor.offset());
- assertFalse(cursor.hasRemaining());
+ assertFalse(cursor.hasNext());
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
