This is an automated email from the ASF dual-hosted git repository.
peterxcli 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 d5d4df4d4b6 HDDS-16395. Add lock-free StreamBlock pread (#11314)
d5d4df4d4b6 is described below
commit d5d4df4d4b625a645ed406b2334fbb15d17342eb
Author: Tak Lon (Stephen) Wu <[email protected]>
AuthorDate: Wed Sep 30 20:37:21 2026 -0700
HDDS-16395. Add lock-free StreamBlock pread (#11314)
---
.../hadoop/hdds/scm/storage/BlockInputStream.java | 6 +-
.../hdds/scm/storage/MultipartInputStream.java | 48 +----
.../hadoop/hdds/scm/storage/PartInputStream.java | 13 ++
.../hdds/scm/storage/StreamBlockInputStream.java | 159 ++++++++++++++---
.../hdds/scm/storage/TestMultipartInputStream.java | 69 +++++++
.../scm/storage/TestStreamBlockInputStream.java | 198 +++++++++++++++++++++
.../hadoop/hdds/utils/ConnectionFailureUtils.java | 2 +-
.../hdds/utils/TestConnectionFailureUtils.java | 14 ++
.../ozone/client/io/OzoneCryptoInputStream.java | 5 +
9 files changed, 439 insertions(+), 75 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java
index 57757cb79f7..c3b1f3ed73f 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java
@@ -473,7 +473,11 @@ protected synchronized int
readWithStrategy(ByteReaderStrategy strategy)
*
* @return bytes copied into {@code dst}, or {@link #EOF} at EOF
*/
- int readPositioned(long blockRelativePosition, ByteBuffer dst) throws
IOException {
+ @Override
+ public int readPositioned(long blockRelativePosition, ByteBuffer dst) throws
IOException {
+ if (!dst.hasRemaining()) {
+ return EOF;
+ }
if (blockRelativePosition < 0) {
return EOF;
}
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java
index 584d97c778e..00a119690b0 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java
@@ -196,58 +196,19 @@ public synchronized void seek(long pos) throws
IOException {
@Override
public boolean readFully(long position, ByteBuffer buffer) throws
IOException {
- if (isStreamBlockInputStream) {
- return readFullyStreamBlock(position, buffer);
+ if (!statelessPositionedReadSupported) {
+ return false;
}
return readFullyStateless(position, buffer);
}
/**
- * Positioned read for the StreamBlock path. This emulates a positioned read
- * with seek-read-restore on the shared stream cursor, so it is synchronized
- * to keep concurrent callers from corrupting each other's position.
- */
- private synchronized boolean readFullyStreamBlock(long position,
- ByteBuffer buffer) throws IOException {
- final long oldPos = getPos();
- seek(position);
- try {
- int remainingBeforeRead = buffer.remaining();
- if (remainingBeforeRead == 0) {
- return true;
- }
- read(new ByteBufferReader(buffer) {
- @Override
- int readImpl(InputStream inputStream) throws IOException {
- return Preconditions.assertInstanceOf(inputStream,
StreamBlockInputStream.class)
- .readFully(getBuffer(), false);
- }
- });
- if (remainingBeforeRead - buffer.remaining() == 0) {
- throw new EOFException("EOF encountered at pos: " + position +
- " for key: " + key);
- }
- } finally {
- seek(oldPos);
- }
- return true;
- }
-
- /**
- * Stateless positioned read for the replicated (Ratis) path. Routes the read
- * across the part {@link BlockInputStream}s using the immutable
- * {@link #partOffsets} without seeking or mutating the shared cursor, so
- * concurrent positioned reads run independently. Returns {@code false} (so
- * the caller can fall back) when any part is not a {@link BlockInputStream},
- * e.g. erasure coded parts.
+ * Shared stateless positioned-read loop across parts using {@link
PartInputStream#readPositioned}.
*/
private boolean readFullyStateless(long position, ByteBuffer buffer) throws
IOException {
if (!buffer.hasRemaining()) {
return true;
}
- if (!statelessPositionedReadSupported) {
- return false;
- }
long pos = position;
int bytesRead = 0;
@@ -260,9 +221,8 @@ private boolean readFullyStateless(long position,
ByteBuffer buffer) throws IOEx
" for key: " + key);
}
int idx = binarySearchOffsetIndex(partOffsets, pos);
- BlockInputStream part = (BlockInputStream) partStreams.get(idx);
long partPos = pos - partOffsets[idx];
- int n = part.readPositioned(partPos, buffer);
+ int n = partStreams.get(idx).readPositioned(partPos, buffer);
if (n <= 0) {
if (bytesRead > 0) {
return true;
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/PartInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/PartInputStream.java
index 7aa0d594c38..3d7ee44c58e 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/PartInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/PartInputStream.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdds.scm.storage;
import java.io.IOException;
+import java.nio.ByteBuffer;
import org.apache.hadoop.fs.CanUnbuffer;
import org.apache.hadoop.fs.Seekable;
@@ -33,4 +34,16 @@ default long getRemaining() throws IOException {
}
void close() throws IOException;
+
+ /**
+ * Positioned read within this part starting at {@code partOffset}.
+ *
+ * @return bytes copied into {@code buffer}, {@code -1} if {@code buffer}
has no remaining space
+ * or at EOF
+ * @throws UnsupportedOperationException if this part stream does not
support positioned read
+ */
+ default int readPositioned(long partOffset, ByteBuffer buffer) throws
IOException {
+ throw new UnsupportedOperationException(
+ "Positioned read is not supported by " + getClass().getName());
+ }
}
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 ddef8e54f42..ce587f37292 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
@@ -23,10 +23,10 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.Duration;
-import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@@ -58,6 +58,7 @@
import org.apache.hadoop.security.token.Token;
import org.apache.ratis.protocol.exceptions.TimeoutIOException;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.io.grpc.Status;
import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException;
import org.apache.ratis.thirdparty.io.grpc.stub.ClientCallStreamObserver;
import org.apache.ratis.util.Preconditions;
@@ -96,7 +97,8 @@ public class StreamBlockInputStream extends
BlockExtendedInputStream {
private final Function<BlockID, BlockLocationInfo> refreshFunction;
private final RetryPolicy retryPolicy;
private int retries = 0;
- private final Set<DatanodeID> failedStreamingDatanodes = new HashSet<>();
+ // Shared by sequential and positioned reads; must be thread-safe for
concurrent access.
+ private final Set<DatanodeID> failedStreamingDatanodes =
ConcurrentHashMap.newKeySet();
public StreamBlockInputStream(
BlockID blockID, long length, Pipeline pipeline,
@@ -178,6 +180,96 @@ synchronized int readFully(ByteBuffer targetBuf, boolean
preRead) throws IOExcep
return read > 0 ? read : EOF;
}
+ /**
+ * Positioned read served by a dedicated one-shot streaming read which never
touches the
+ * cursor, the pre-read buffer or the sequential reader.
+ *
+ * @param blockOffset the offset within the block to read from.
+ * @param dst the buffer to read into.
+ * @return the number of bytes copied into {@code dst}, or -1 if no byte
could be read.
+ */
+ @Override
+ public int readPositioned(long blockOffset, ByteBuffer dst) throws
IOException {
+ if (!dst.hasRemaining()) {
+ return EOF;
+ }
+ if (blockOffset < 0 || blockOffset >= blockLength) {
+ return EOF;
+ }
+ final XceiverClientFactory factory;
+ synchronized (this) {
+ checkOpen();
+ factory = xceiverClientFactory;
+ }
+ final int length = Math.toIntExact(Math.min(dst.remaining(), blockLength -
blockOffset));
+ final int startPosition = dst.position();
+ int callRetries = 0;
+ while (true) {
+ final Pipeline pipeline = pipelineRef.get();
+ final XceiverClientGrpc client = acquireGrpcClientForReadData(factory,
pipeline);
+ final StreamingReader reader = new StreamingReader(client);
+ try {
+ // On failure initStreamRead releases its own permit, so the reader is
closed only after it succeeds.
+ client.initStreamRead(blockID, reader, failedStreamingDatanodes);
+ try {
+ return preadOnce(client, reader, pipeline, blockOffset, length, dst);
+ } finally {
+ closeStream(reader);
+ }
+ } catch (IOException e) {
+ dst.position(startPosition);
+ handlePreadException(e, reader, blockOffset, callRetries++);
+ } finally {
+ factory.releaseClientForReadData(client, false);
+ }
+ }
+ }
+
+ private int preadOnce(XceiverClientGrpc client, StreamingReader reader,
Pipeline pipeline,
+ long blockOffset, int length, ByteBuffer dst) throws IOException {
+ final StreamingReadResponse response = reader.getResponse();
+ if (response == null) {
+ throw new IOException("Uninitialized StreamingReadResponse: " + blockID);
+ }
+ client.streamRead(ContainerProtocolCalls.buildReadBlockCommandProto(
+ blockID, blockOffset, length, responseDataSize, tokenRef.get(),
pipeline), response);
+
+ int copied = 0;
+ while (copied < length) {
+ final ReadBlockResponseProto proto = reader.poll();
+ if (proto == null) {
+ break;
+ }
+ final ByteBuffer buffer = getByteBuffer(proto, blockOffset + copied);
+ if (buffer == null || !buffer.hasRemaining()) {
+ continue;
+ }
+ buffer.limit(buffer.position() + Math.min(buffer.remaining(), length -
copied));
+ copied += buffer.remaining();
+ dst.put(buffer);
+ }
+ return copied > 0 ? copied : EOF;
+ }
+
+ private void handlePreadException(IOException cause, StreamingReader reader,
long blockOffset,
+ int callRetries) throws IOException {
+ final IOException root = ConnectionFailureUtils.unwrapCause(cause);
+ if (Status.fromThrowable(cause).getCode() == Status.Code.OUT_OF_RANGE) {
+ final EOFException eof = new EOFException("Failed to read block " +
blockID + " at offset "
+ + blockOffset + ": " + cause.getMessage());
+ eof.initCause(cause);
+ throw eof;
+ }
+ final boolean retriable = root instanceof StorageContainerException ||
isConnectivityIssue(root)
+ || root instanceof TimeoutIOException;
+ if (!retriable || !shouldRetryRead(root, retryPolicy, callRetries)) {
+ throw cause;
+ }
+ recordFailedStreamingDatanode(reader);
+ refreshBlockInfo(root);
+ LOG.warn("Refreshing block data to pread block {} due to {}", blockID,
cause.getMessage());
+ }
+
private synchronized boolean dataAvailableToRead(int length, boolean
preRead) throws IOException {
if (position >= blockLength) {
return false;
@@ -285,7 +377,10 @@ private synchronized void closeReader(String reason) {
final StreamingReader reader = streamingReader;
streamingReader = null;
LOG.debug("{} closeReader for {}", getName(reader), reason);
+ closeStream(reader);
+ }
+ private static void closeStream(StreamingReader reader) {
reader.onCompleted();
final StreamingReadResponse response = reader.getResponse();
@@ -314,31 +409,34 @@ protected synchronized void checkOpen() throws
IOException {
protected synchronized void acquireClient() throws IOException {
checkOpen();
if (xceiverClient == null) {
- final Pipeline pipeline = pipelineRef.get();
- final XceiverClientSpi client;
- try {
- client = xceiverClientFactory.acquireClientForReadData(pipeline);
- } catch (IOException ioe) {
- LOG.warn("Failed to acquire client for pipeline {}, block {}",
pipeline, blockID);
- throw ioe;
- }
-
- if (client == null) {
- throw new IOException("Failed to acquire client for " + pipeline);
- }
- if (!(client instanceof XceiverClientGrpc)) {
- throw new IOException("Unexpected client class: " +
client.getClass().getName() + ", " + pipeline);
- }
+ xceiverClient = acquireGrpcClientForReadData(xceiverClientFactory,
pipelineRef.get());
+ }
+ }
- xceiverClient = (XceiverClientGrpc) client;
+ private XceiverClientGrpc acquireGrpcClientForReadData(XceiverClientFactory
factory, Pipeline pipeline)
+ throws IOException {
+ final XceiverClientSpi acquired;
+ try {
+ acquired = factory.acquireClientForReadData(pipeline);
+ } catch (IOException ioe) {
+ LOG.warn("Failed to acquire client for pipeline {}, block {}", pipeline,
blockID);
+ throw ioe;
+ }
+ if (acquired == null) {
+ throw new IOException("Failed to acquire client for " + pipeline);
+ }
+ if (!(acquired instanceof XceiverClientGrpc)) {
+ factory.releaseClientForReadData(acquired, false);
+ throw new IOException("Unexpected client class: " +
acquired.getClass().getName() + ", " + pipeline);
}
+ return (XceiverClientGrpc) acquired;
}
private synchronized void initialize() throws IOException {
while (streamingReader == null) {
try {
acquireClient();
- final StreamingReader reader = new StreamingReader();
+ final StreamingReader reader = new StreamingReader(xceiverClient);
LOG.debug("{}: new StreamingReader", getName(reader));
xceiverClient.initStreamRead(blockID, reader,
failedStreamingDatanodes);
streamingReader = reader;
@@ -394,10 +492,14 @@ private void handleExceptions(IOException cause) throws
IOException {
}
private void recordFailedStreamingDatanode() {
- if (streamingReader == null) {
+ recordFailedStreamingDatanode(streamingReader);
+ }
+
+ private void recordFailedStreamingDatanode(StreamingReader reader) {
+ if (reader == null) {
return;
}
- final StreamingReadResponse response = streamingReader.getResponse();
+ final StreamingReadResponse response = reader.getResponse();
if (response == null) {
return;
}
@@ -426,12 +528,6 @@ private void refreshBlockInfo(IOException cause) throws
IOException {
refreshBlockInfo(cause, blockID, pipelineRef, tokenRef, refreshFunction);
}
- private synchronized void releaseStreamResources() {
- if (xceiverClient != null) {
- xceiverClient.completeStreamRead();
- }
- }
-
@Override
public String toString() {
return name;
@@ -480,10 +576,11 @@ public String toString() {
}
/**
- * Implementation of a StreamObserver used to received and buffer streaming
GRPC reads.
+ * StreamObserver used for sequential reads and for one-shot positioned
reads.
*/
public class StreamingReader implements StreamingReaderSpi {
private final String name = StreamBlockInputStream.this.name + "-reader" +
READER_ID.getAndIncrement();
+ private final XceiverClientGrpc client;
/** Response queue: poll is blocking while offer is non-blocking. */
private final BlockingQueue<ReadBlockResponseProto> responseQueue = new
LinkedBlockingQueue<>();
@@ -492,6 +589,10 @@ public class StreamingReader implements StreamingReaderSpi
{
private final AtomicBoolean semaphoreReleased = new AtomicBoolean(false);
private final AtomicReference<StreamingReadResponse> response = new
AtomicReference<>();
+ StreamingReader(XceiverClientGrpc client) {
+ this.client = client;
+ }
+
void checkError() throws IOException {
if (future.isCompletedExceptionally()) {
try {
@@ -569,7 +670,7 @@ private ReadBuffer read(int length, boolean preRead) throws
IOException {
private void releaseResources() {
if (semaphoreReleased.compareAndSet(false, true)) {
- releaseStreamResources();
+ client.completeStreamRead();
}
}
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestMultipartInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestMultipartInputStream.java
index 59a010c127f..7ac64376390 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestMultipartInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestMultipartInputStream.java
@@ -21,7 +21,13 @@
import static
org.apache.hadoop.hdds.scm.storage.TestChunkInputStream.generateRandomData;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import com.google.common.primitives.Bytes;
import java.nio.ByteBuffer;
@@ -41,6 +47,7 @@
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.ozone.common.Checksum;
import org.junit.jupiter.api.Test;
+import org.mockito.invocation.InvocationOnMock;
/**
* Tests for {@link MultipartInputStream}'s functionality.
@@ -73,6 +80,49 @@ public void testPositionedReadPartialAtEof() throws
Exception {
}
}
+ @Test
+ public void testStreamBlockPositionedReadPartialAtEof() throws Exception {
+ int fileLen = 1024;
+ byte[] partData = generateRandomData(fileLen);
+ StreamBlockInputStream part = stubStreamBlockPart(partData);
+ try (MultipartInputStream multipartStream =
+ new MultipartInputStream("test-key", Collections.singletonList(part)))
{
+ multipartStream.initialize();
+ assertTrue(multipartStream.isStreamBlockInputStream());
+ int position = 12;
+ int expectedBytes = fileLen - position;
+ ByteBuffer buffer = ByteBuffer.allocate(expectedBytes * 2);
+ assertTrue(multipartStream.readFully(position, buffer));
+ assertEquals(expectedBytes, buffer.position());
+ verify(part, never()).seek(anyLong());
+ }
+ }
+
+ @Test
+ public void testStreamBlockConcurrentPositionedRead() throws Exception {
+ byte[] part0Data = generateRandomData(PART_SIZE);
+ byte[] part1Data = generateRandomData(PART_SIZE);
+ byte[] keyData = Bytes.concat(part0Data, part1Data);
+
+ StreamBlockInputStream part0 = stubStreamBlockPart(part0Data);
+ StreamBlockInputStream part1 = stubStreamBlockPart(part1Data);
+ List<StreamBlockInputStream> parts = new ArrayList<>();
+ parts.add(part0);
+ parts.add(part1);
+ try (MultipartInputStream multipartStream = new
MultipartInputStream("test-key", parts)) {
+ multipartStream.initialize();
+ assertTrue(multipartStream.isStreamBlockInputStream());
+ PositionedReadTestHelper.runConcurrentPositionedReads(keyData,
+ (offset, buf) -> {
+ if (!multipartStream.readFully(offset, buf)) {
+ throw new AssertionError("stateless readFully returned false at
" + offset);
+ }
+ });
+ verify(part0, never()).seek(anyLong());
+ verify(part1, never()).seek(anyLong());
+ }
+ }
+
@Test
public void testConcurrentPositionedRead() throws Exception {
byte[] part0Data = generateRandomData(PART_SIZE);
@@ -105,6 +155,25 @@ public void testConcurrentPositionedRead() throws
Exception {
}
}
+ private static StreamBlockInputStream stubStreamBlockPart(byte[] partData)
throws Exception {
+ StreamBlockInputStream part = mock(StreamBlockInputStream.class);
+ when(part.getLength()).thenReturn((long) partData.length);
+ doAnswer(invocation -> readPositionedFrom(partData, invocation)).when(part)
+ .readPositioned(anyLong(), any(ByteBuffer.class));
+ return part;
+ }
+
+ private static int readPositionedFrom(byte[] data, InvocationOnMock
invocation) {
+ long offset = invocation.getArgument(0);
+ ByteBuffer dst = invocation.getArgument(1);
+ if (offset < 0 || offset >= data.length) {
+ return -1;
+ }
+ int toCopy = Math.min(dst.remaining(), data.length - (int) offset);
+ dst.put(data, (int) offset, toCopy);
+ return toCopy;
+ }
+
private static BlockInputStream createBlockStream(BlockID blockID, byte[]
blockData,
Pipeline pipeline, Function<BlockID, BlockLocationInfo> refreshFunction,
OzoneClientConfig clientConfig, Checksum checksum) throws Exception {
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 98398326381..88258bfe547 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
@@ -17,12 +17,15 @@
package org.apache.hadoop.hdds.scm.storage;
+import static
org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.BUFFER_SIZE;
+import static
org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
@@ -36,7 +39,10 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.Duration;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
@@ -107,6 +113,87 @@ public void testCustomStreamReadConfigIsApplied() throws
Exception {
}
}
+ @Test
+ public void testPositionedReadIsStateless() throws Exception {
+ byte[] data = new byte[100];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte) i;
+ }
+ try (StreamBlockInputStream sbis = newStreamForData(data)) {
+ byte[] seq = new byte[10];
+ assertEquals(10, sbis.read(seq, 0, 10));
+ assertArrayEquals(Arrays.copyOfRange(data, 0, 10), seq);
+ long posBeforePositionedRead = sbis.getPos();
+ assertEquals(10, posBeforePositionedRead);
+
+ ByteBuffer dst = ByteBuffer.allocate(15);
+ int read = sbis.readPositioned(50, dst);
+ assertEquals(15, read);
+ dst.flip();
+ byte[] positioned = new byte[15];
+ dst.get(positioned);
+ assertArrayEquals(Arrays.copyOfRange(data, 50, 65), positioned);
+ assertEquals(posBeforePositionedRead, sbis.getPos());
+
+ byte[] seq2 = new byte[10];
+ assertEquals(10, sbis.read(seq2, 0, 10));
+ assertArrayEquals(Arrays.copyOfRange(data, 10, 20), seq2);
+ }
+ }
+
+ @Test
+ public void testPositionedReadAtEof() throws Exception {
+ byte[] data = new byte[100];
+ try (StreamBlockInputStream sbis = newStreamForData(data)) {
+ ByteBuffer dst = ByteBuffer.allocate(10);
+ assertEquals(-1, sbis.readPositioned(data.length, dst));
+ assertEquals(-1, sbis.readPositioned(data.length + 1, dst));
+ // An empty buffer also gets -1, as PartInputStream#readPositioned
documents.
+ assertEquals(-1, sbis.readPositioned(0, ByteBuffer.allocate(0)));
+ }
+ }
+
+ @Test
+ public void testPositionedReadDoesNotReleasePermitWhenInitStreamReadFails()
throws Exception {
+ OzoneClientConfig clientConfig = newStreamReadConfig();
+ clientConfig.setMaxReadRetryCount(0);
+ BlockID blockID = new BlockID(1L, 18L);
+ Pipeline pipeline = mockStandalonePipeline();
+
+ // Like XceiverClientGrpc, fail before the gRPC call starts:
initStreamRead then releases its own permit.
+ XceiverClientGrpc xceiverClient = mock(XceiverClientGrpc.class);
+ doThrow(new IOException("initStreamRead failed"))
+ .when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
+
+ XceiverClientFactory xceiverClientFactory =
mock(XceiverClientFactory.class);
+ when(xceiverClientFactory.acquireClientForReadData(any(Pipeline.class)))
+ .thenReturn(xceiverClient);
+ doNothing().when(xceiverClientFactory).releaseClientForReadData(any(),
anyBoolean());
+
+ try (StreamBlockInputStream sbis = new StreamBlockInputStream(
+ blockID, 64L, pipeline, null, xceiverClientFactory, NO_REFRESH,
clientConfig)) {
+ ByteBuffer dst = ByteBuffer.allocate(16);
+ assertThrows(IOException.class, () -> sbis.readPositioned(0, dst));
+ }
+
+ verify(xceiverClient, never()).completeStreamRead();
+ }
+
+ @Test
+ public void testConcurrentPositionedRead() throws Exception {
+ byte[] data = TestChunkInputStream.generateRandomData(SOURCE_SIZE);
+ try (StreamBlockInputStream sbis = newStreamForData(data)) {
+ PositionedReadTestHelper.runConcurrentPositionedReads(data,
+ (offset, buf) -> {
+ int read = sbis.readPositioned(offset, buf);
+ if (read != BUFFER_SIZE) {
+ throw new AssertionError("expected " + BUFFER_SIZE + " bytes at
" + offset
+ + " but read " + read);
+ }
+ });
+ }
+ }
+
@Test
public void testReleasesStreamPermitAtBlockEof() throws Exception {
OzoneClientConfig clientConfig = newStreamReadConfig();
@@ -270,6 +357,55 @@ public void
testPollDoesNotDropQueuedItemWhenFutureCompletesFirst() throws Excep
}
}
+ private StreamBlockInputStream newStreamForData(byte[] data) throws
Exception {
+ BlockID blockID = new BlockID(1L, 17L);
+ Pipeline pipeline = mockStandalonePipeline();
+ ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver =
+ mock(ClientCallStreamObserver.class);
+ XceiverClientGrpc xceiverClient = mockOffsetAwareStreamingReadClient(data,
requestObserver);
+ XceiverClientFactory xceiverClientFactory =
mock(XceiverClientFactory.class);
+ when(xceiverClientFactory.acquireClientForReadData(any(Pipeline.class)))
+ .thenReturn(xceiverClient);
+ doNothing().when(xceiverClientFactory).releaseClientForReadData(any(),
anyBoolean());
+ return new StreamBlockInputStream(blockID, data.length, pipeline, null,
+ xceiverClientFactory, NO_REFRESH, newStreamReadConfig());
+ }
+
+ private XceiverClientGrpc mockOffsetAwareStreamingReadClient(byte[] data,
+ ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver)
throws Exception {
+ XceiverClientGrpc xceiverClient = mock(XceiverClientGrpc.class);
+ Map<StreamingReadResponse, StreamingReaderSpi> readersByResponse = new
ConcurrentHashMap<>();
+ doAnswer(invocation -> {
+ StreamingReaderSpi reader = invocation.getArgument(1);
+ StreamingReadResponse response = mock(StreamingReadResponse.class);
+ when(response.getRequestObserver()).thenReturn(requestObserver);
+ reader.setStreamingReadResponse(response);
+ readersByResponse.put(response, reader);
+ return null;
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
+ doAnswer(invocation -> {
+ ContainerCommandRequestProto request = invocation.getArgument(0);
+ StreamingReadResponse response = invocation.getArgument(1);
+ StreamingReaderSpi reader = readersByResponse.get(response);
+ long offset = request.getReadBlock().getOffset();
+ long length = request.getReadBlock().getLength();
+ int start = (int) offset;
+ int end = (int) Math.min(data.length, offset + length);
+ if (start >= data.length) {
+ reader.onCompleted();
+ return null;
+ }
+ byte[] slice = Arrays.copyOfRange(data, start, end);
+ reader.onNext(buildResponseProto(slice, offset));
+ if (end >= data.length) {
+ reader.onCompleted();
+ }
+ return null;
+ }).when(xceiverClient).streamRead(any(ContainerCommandRequestProto.class),
+ any(StreamingReadResponse.class));
+ return xceiverClient;
+ }
+
private OzoneClientConfig newStreamReadConfig() {
OzoneClientConfig clientConfig = new OzoneClientConfig();
clientConfig.setChecksumVerify(false);
@@ -822,6 +958,68 @@ public void testFailFastOnGrpcOutOfRangeStatus() throws
Exception {
}
}
+ /**
+ * A streamed error response (e.g. CONTAINER_NOT_FOUND) is retried after
refreshing the block location,
+ * for both sequential and positioned reads.
+ */
+ @Test
+ public void testRetriesErrorResponseAfterRefresh() throws Exception {
+ OzoneClientConfig clientConfig = newStreamReadConfig();
+ clientConfig.setMaxReadRetryCount(1);
+ clientConfig.setReadRetryInterval(0);
+ byte[] data = new byte[] {1, 2, 3, 4};
+ ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver =
mock(ClientCallStreamObserver.class);
+ StreamingReadResponse streamingReadResponse =
mock(StreamingReadResponse.class);
+
when(streamingReadResponse.getRequestObserver()).thenReturn(requestObserver);
+
when(streamingReadResponse.getDatanodeDetails()).thenReturn(MockDatanodeDetails.randomDatanodeDetails());
+
+ // Every other ReadBlock request gets an error response; only one reader
is active at a time here.
+ AtomicInteger streamReads = new AtomicInteger();
+ AtomicReference<StreamingReaderSpi> lastReader = new AtomicReference<>();
+ XceiverClientGrpc xceiverClient = mock(XceiverClientGrpc.class);
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = inv.getArgument(1);
+ reader.setStreamingReadResponse(streamingReadResponse);
+ lastReader.set(reader);
+ return null;
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = lastReader.get();
+ if (streamReads.getAndIncrement() % 2 == 0) {
+ reader.onNext(ContainerCommandResponseProto.newBuilder()
+ .setCmdType(Type.ReadBlock)
+ .setResult(ContainerProtos.Result.CONTAINER_NOT_FOUND)
+ .setMessage("Container not found")
+ .build());
+ } else {
+ reader.onNext(buildResponseProto(data, 0));
+ reader.onCompleted();
+ }
+ return null;
+ }).when(xceiverClient).streamRead(any(), any());
+ XceiverClientFactory xceiverClientFactory =
mock(XceiverClientFactory.class);
+
when(xceiverClientFactory.acquireClientForReadData(any(Pipeline.class))).thenReturn(xceiverClient);
+ AtomicInteger refreshes = new AtomicInteger();
+ Function<BlockID, BlockLocationInfo> refresh = b -> {
+ refreshes.incrementAndGet();
+ return null;
+ };
+
+ try (StreamBlockInputStream sbis = new StreamBlockInputStream(
+ new BlockID(1L, 22L), data.length, mockStandalonePipeline(), null,
xceiverClientFactory,
+ refresh, clientConfig)) {
+ ByteBuffer positioned = ByteBuffer.allocate(data.length);
+ assertEquals(data.length, sbis.readPositioned(0, positioned));
+ assertArrayEquals(data, positioned.array());
+ assertEquals(1, refreshes.get());
+
+ ByteBuffer sequential = ByteBuffer.allocate(data.length);
+ assertEquals(data.length, sbis.read(sequential));
+ assertArrayEquals(data, sequential.array());
+ assertEquals(2, refreshes.get());
+ }
+ }
+
/**
* Mocks a streaming read client which captures the StreamingReaderSpi during
* initStreamRead and drives the given callback when a ReadBlock request is
sent.
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java
index d7f45b1cd7e..9ba515e91a2 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java
@@ -120,7 +120,7 @@ public static IOException unwrapCause(IOException ex) {
t = t.getCause();
continue;
}
- if (t.getCause() instanceof IOException) {
+ if (t.getCause() instanceof IOException || t.getCause() instanceof
ExecutionException) {
t = t.getCause();
continue;
}
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/utils/TestConnectionFailureUtils.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/utils/TestConnectionFailureUtils.java
index 00c327f6899..ca8f46570be 100644
---
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/utils/TestConnectionFailureUtils.java
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/utils/TestConnectionFailureUtils.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdds.utils;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.EOFException;
@@ -27,7 +28,10 @@
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
+import java.util.concurrent.ExecutionException;
import java.util.stream.Stream;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result;
+import
org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.security.AccessControlException;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -117,6 +121,16 @@ public void testNullIsNotAConnectionFailure() {
assertFalse(ConnectionFailureUtils.isConnectionFailure(null));
}
+ /**
+ * The shape StreamBlockInputStream's reader throws for a failed stream.
+ */
+ @Test
+ public void testUnwrapCauseThroughIOExceptionWrappingExecutionException() {
+ StorageContainerException sce = new StorageContainerException("not found",
Result.CONTAINER_NOT_FOUND);
+ IOException wrapped = new IOException("Streaming read failed", new
ExecutionException(sce));
+ assertSame(sce, ConnectionFailureUtils.unwrapCause(wrapped));
+ }
+
/**
* {@code Throwable.initCause} contractually rejects setting cause to
* the throwable itself, but cycles of length 2+ have appeared in
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneCryptoInputStream.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneCryptoInputStream.java
index ab8617e30f4..705aa84d637 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneCryptoInputStream.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneCryptoInputStream.java
@@ -188,6 +188,11 @@ public synchronized int available() throws IOException {
return super.available();
}
+ @Override
+ public synchronized int readPositioned(long partOffset, ByteBuffer buffer)
throws IOException {
+ return read(partOffset, buffer);
+ }
+
/**
* Positioned read. Decryption can only happen at Crypto buffer boundaries,
so this stream cannot read
* at an arbitrary position without moving its cursor. The read is
serialized against the other reads:
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
