This is an automated email from the ASF dual-hosted git repository.
szetszwo 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 63f93f7ca79 HDDS-16323. StreamBlockInputStream: refill the pre-read
window in bulk instead of once per response (#11159)
63f93f7ca79 is described below
commit 63f93f7ca79b37312dd07bbcc98b3d8fca25eea8
Author: Sergey Soldatov <[email protected]>
AuthorDate: Fri Oct 2 08:58:08 2026 -0700
HDDS-16323. StreamBlockInputStream: refill the pre-read window in bulk
instead of once per response (#11159)
---
.../apache/hadoop/hdds/scm/OzoneClientConfig.java | 4 +-
.../hdds/scm/storage/StreamBlockInputStream.java | 12 +++-
.../scm/storage/TestStreamBlockInputStream.java | 78 ++++++++++++++++++++++
3 files changed, 92 insertions(+), 2 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java
index 824f2674f55..493d11f37be 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java
@@ -338,7 +338,9 @@ public class OzoneClientConfig {
defaultValue = "33554432",
type = ConfigType.LONG,
tags = {ConfigTag.CLIENT},
- description = "Extra bytes to prefetch during streaming reads.")
+ description = "Size of the pre-read window for streaming reads: the
number of bytes requested from the " +
+ "datanode ahead of the current read position. The window is refilled
in bulk once the bytes requested " +
+ "ahead drop below half of this value. Set to 0 to disable pre-read.")
private long streamReadPreReadSize = 32L << 20;
@Config(key = "ozone.client.stream.read.response-data-size",
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
index ce587f37292..d773701300e 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
@@ -447,8 +447,13 @@ private synchronized void initialize() throws IOException {
}
synchronized void readBlock(int length, boolean preRead) throws IOException {
- final long required = position + length - requestedLength;
+ final long outstanding = requestedLength - position; // bytes requested
from the datanode but not yet consumed
+ final long required = length - outstanding; // bytes required to
fulfill this call
final long preReadLength = preRead ? preReadSize : 0;
+ if (required <= 0 && outstanding >=
getPreReadRefillThreshold(preReadLength)) {
+ // Safe to skip: already has the required bytes and exceeded the refill
threshold
+ return;
+ }
// Clamp so requestedLength never exceeds blockLength: requesting past the
end
// produces an offset the DataNode cannot serve, causing a read timeout.
final long readLength = Math.min(required + preReadLength, blockLength -
requestedLength);
@@ -537,6 +542,11 @@ public long getPreReadSize() {
return preReadSize;
}
+ /** Refill the pre-read window in bulk once it drains below this, instead of
after every response. */
+ static long getPreReadRefillThreshold(long preRead) {
+ return preRead / 2;
+ }
+
public int getResponseDataSize() {
return responseDataSize;
}
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
index 88258bfe547..d162b80ad94 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
@@ -39,8 +39,10 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
@@ -406,6 +408,82 @@ private XceiverClientGrpc
mockOffsetAwareStreamingReadClient(byte[] data,
return xceiverClient;
}
+ /**
+ * The pre-read window must be refilled in bulk once it drains below half,
not after every response.
+ */
+ @Test
+ public void testPreReadWindowIsRefilledInBulk() throws Exception {
+ final int responseSize = 4096;
+ final int preReadSize = 4 * responseSize;
+ final int blockLength = 16 * responseSize;
+
+ OzoneClientConfig clientConfig = newStreamReadConfig();
+ clientConfig.setStreamReadResponseDataSize(responseSize);
+ clientConfig.setStreamReadPreReadSize(preReadSize);
+
+ BlockID blockID = new BlockID(1L, 21L);
+ byte[] data = new byte[blockLength];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte) i;
+ }
+ ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver =
mock(ClientCallStreamObserver.class);
+ StreamingReadResponse streamingReadResponse = new StreamingReadResponse(
+ MockDatanodeDetails.randomDatanodeDetails(), requestObserver);
+
+ AtomicReference<StreamingReaderSpi> readerRef = new AtomicReference<>();
+ List<Long> requestLengths = Collections.synchronizedList(new
ArrayList<>());
+
+ XceiverClientGrpc xceiverClient = mock(XceiverClientGrpc.class);
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = inv.getArgument(1);
+ reader.setStreamingReadResponse(streamingReadResponse);
+ readerRef.set(reader);
+ return null;
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
+
+ doAnswer(inv -> {
+ ContainerCommandRequestProto request = inv.getArgument(0);
+ final long offset = request.getReadBlock().getOffset();
+ final long length = request.getReadBlock().getLength();
+ requestLengths.add(length);
+ for (long sent = 0; sent < length; sent += responseSize) {
+ final int from = Math.toIntExact(offset + sent);
+ final int to = Math.toIntExact(Math.min((long) from + responseSize,
blockLength));
+ readerRef.get().onNext(buildResponseProto(Arrays.copyOfRange(data,
from, to), from));
+ }
+ return null;
+ }).when(xceiverClient).streamRead(any(), any());
+
+ XceiverClientFactory xceiverClientFactory =
mock(XceiverClientFactory.class);
+ when(xceiverClientFactory.acquireClientForReadData(any(Pipeline.class)))
+ .thenReturn(xceiverClient);
+
+ final long refillThreshold;
+ try (StreamBlockInputStream sbis = new StreamBlockInputStream(
+ blockID, blockLength, mockStandalonePipeline(), null,
xceiverClientFactory,
+ NO_REFRESH, clientConfig)) {
+ refillThreshold =
StreamBlockInputStream.getPreReadRefillThreshold(sbis.getPreReadSize());
+ // Read one small buffer at a time, as KeyInputStream does, so every
response reaches readBlock again.
+ byte[] all = new byte[blockLength];
+ for (int off = 0; off < blockLength; off += responseSize) {
+ ByteBuffer chunk = ByteBuffer.allocate(responseSize);
+ assertEquals(responseSize, sbis.read(chunk));
+ System.arraycopy(chunk.array(), 0, all, off, responseSize);
+ }
+ assertArrayEquals(data, all);
+ }
+
+ // One request per refill threshold plus the initial one; without the gate
it is one per response.
+ assertThat(requestLengths)
+ .withFailMessage("expected bulk refills, got %s requests for %s
responses: %s",
+ requestLengths.size(), blockLength / responseSize, requestLengths)
+ .hasSizeLessThanOrEqualTo(Math.toIntExact(1 + blockLength /
refillThreshold));
+ // Each request carries at least the refill threshold (the last one may be
clamped by the block end).
+ for (long requested : requestLengths) {
+ assertThat(requested).isGreaterThanOrEqualTo(refillThreshold);
+ }
+ }
+
private OzoneClientConfig newStreamReadConfig() {
OzoneClientConfig clientConfig = new OzoneClientConfig();
clientConfig.setChecksumVerify(false);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
