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 86e97cd3750 HDDS-16396. Support concurrent positioned reads for EC and
encrypted keys (#11348)
86e97cd3750 is described below
commit 86e97cd3750eb451e4cd26e20e62af474f537ede
Author: Peter Lee <[email protected]>
AuthorDate: Mon Oct 12 18:42:15 2026 +0800
HDDS-16396. Support concurrent positioned reads for EC and encrypted keys
(#11348)
---
.../hadoop/hdds/scm/storage/BlockInputStream.java | 2 +-
.../hadoop/hdds/scm/storage/ChunkInputStream.java | 6 +-
.../hdds/scm/storage/ExtendedInputStream.java | 55 +++-
.../hdds/scm/storage/MultipartInputStream.java | 60 +---
.../hadoop/hdds/scm/storage/PartInputStream.java | 17 +-
.../hdds/scm/storage/StreamBlockInputStream.java | 4 +-
.../hadoop/ozone/client/io/ECBlockInputStream.java | 12 +
.../ozone/client/io/ECBlockInputStreamProxy.java | 39 ++-
.../client/io/ECBlockReconstructedInputStream.java | 12 +
.../hdds/scm/storage/PositionedReadTestHelper.java | 3 +-
.../hdds/scm/storage/TestBlockInputStream.java | 11 +
.../hdds/scm/storage/TestChunkInputStream.java | 3 +
.../hdds/scm/storage/TestMultipartInputStream.java | 54 ++--
.../scm/storage/TestStreamBlockInputStream.java | 117 +++++++-
.../hadoop/ozone/client/io/ECStreamTestUtil.java | 15 +-
.../ozone/client/io/TestECBlockInputStream.java | 12 +
.../client/io/TestECBlockInputStreamProxy.java | 137 +++++++++
.../io/TestECBlockReconstructedInputStream.java | 39 +++
.../ozone/client/io/OzoneCryptoInputStream.java | 91 ++----
.../hadoop/ozone/client/io/OzoneInputStream.java | 39 ++-
.../client/io/TestOzoneCryptoInputStream.java | 88 +++++-
hadoop-ozone/integration-test/pom.xml | 6 +
.../hadoop/fs/ozone/TestOzoneFSInputStream.java | 38 +++
.../client/rpc/TestOzoneAtRestEncryption.java | 26 ++
.../ozone/client/rpc/read/TestStreamRead.java | 9 +-
.../hadoop/fs/ozone/CapableOzoneFSInputStream.java | 4 +-
.../apache/hadoop/fs/ozone/OzoneFSInputStream.java | 147 ++--------
.../hadoop/fs/ozone/TestOzoneFSInputStream.java | 325 ++++++++-------------
28 files changed, 843 insertions(+), 528 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 c3b1f3ed73f..d29fe986c65 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
@@ -476,7 +476,7 @@ protected synchronized int
readWithStrategy(ByteReaderStrategy strategy)
@Override
public int readPositioned(long blockRelativePosition, ByteBuffer dst) throws
IOException {
if (!dst.hasRemaining()) {
- return EOF;
+ return 0;
}
if (blockRelativePosition < 0) {
return EOF;
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
index 67f769f56e4..aa811a5734a 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
@@ -446,13 +446,13 @@ protected void readChunkDataIntoBuffers(ChunkInfo
readChunkInfo)
* @return number of bytes copied into {@code dst}, or {@link #EOF} at EOF
*/
int readPositioned(long chunkRelativePosition, ByteBuffer dst) throws
IOException {
+ if (!dst.hasRemaining()) {
+ return 0;
+ }
if (chunkRelativePosition < 0 || chunkRelativePosition >= length) {
return EOF;
}
final int toRead = (int) Math.min(dst.remaining(), length -
chunkRelativePosition);
- if (toRead == 0) {
- return 0;
- }
final ChunkInfo readChunkInfo = getChunkInfo(chunkRelativePosition,
toRead);
final long adjustedOffset = readChunkInfo.getOffset() -
chunkInfo.getOffset();
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java
index 75e483ad55e..2a61c40608d 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java
@@ -17,12 +17,15 @@
package org.apache.hadoop.hdds.scm.storage;
+import java.io.EOFException;
import java.io.IOException;
-import java.io.InputStream;
import java.nio.ByteBuffer;
+import java.nio.ReadOnlyBufferException;
import org.apache.commons.lang3.NotImplementedException;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
import org.apache.hadoop.fs.ByteBufferReadable;
import org.apache.hadoop.fs.CanUnbuffer;
+import org.apache.hadoop.fs.FSInputStream;
import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.fs.StreamCapabilities;
import org.apache.hadoop.util.StringUtils;
@@ -31,20 +34,51 @@
* Abstact class which extends InputStream and some common interfaces used by
* various Ozone InputStream classes.
*/
-public abstract class ExtendedInputStream extends InputStream
- implements Seekable, CanUnbuffer, ByteBufferReadable, StreamCapabilities {
+public abstract class ExtendedInputStream extends FSInputStream
+ implements Seekable, CanUnbuffer, ByteBufferReadable,
ByteBufferPositionedReadable, StreamCapabilities {
protected static final int EOF = -1;
+ @Override
+ public int read(long position, ByteBuffer buffer) throws IOException {
+ if (position < 0) {
+ throw new EOFException("position is negative: " + position);
+ }
+ if (buffer.isReadOnly()) {
+ throw new ReadOnlyBufferException();
+ }
+ return buffer.hasRemaining() ? readPositioned(position, buffer) : 0;
+ }
+
/**
- * Positioned read.
- *
- * @param position the starting position of the read.
- * @param buffer the buffer for storing the data.
- * @return true iff positioned read is supported in this implementation.
+ * Read without changing the sequential cursor. Implementations must allow
concurrent calls.
*/
- public boolean readFully(long position, ByteBuffer buffer) throws
IOException {
- return false;
+ protected abstract int readPositioned(long position, ByteBuffer buffer)
throws IOException;
+
+ @Override
+ public void readFully(long position, ByteBuffer buffer) throws IOException {
+ do {
+ int n = read(position, buffer);
+ if (n < 0) {
+ throw new EOFException("End of stream at position " + position);
+ }
+ if (n == 0 && buffer.hasRemaining()) {
+ throw new IOException("No progress reading at position " + position);
+ }
+ position += n;
+ } while (buffer.hasRemaining());
+ }
+
+ @Override
+ public int read(long position, byte[] buffer, int offset, int length) throws
IOException {
+ validatePositionedReadArgs(position, buffer, offset, length);
+ return read(position, ByteBuffer.wrap(buffer, offset, length));
+ }
+
+ @Override
+ public void readFully(long position, byte[] buffer, int offset, int length)
throws IOException {
+ validatePositionedReadArgs(position, buffer, offset, length);
+ readFully(position, ByteBuffer.wrap(buffer, offset, length));
}
@Override
@@ -100,6 +134,7 @@ public synchronized boolean seekToNewSource(long l) throws
IOException {
@Override
public boolean hasCapability(String capability) {
switch (StringUtils.toLowerCase(capability)) {
+ case StreamCapabilities.PREADBYTEBUFFER:
case StreamCapabilities.READBYTEBUFFER:
case StreamCapabilities.UNBUFFER:
return true;
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 00a119690b0..d62c1bb42f7 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
@@ -27,7 +27,6 @@
import java.util.List;
import java.util.Objects;
import org.apache.hadoop.fs.FSExceptionMessages;
-import org.apache.ratis.util.Preconditions;
/**
* A stream for accessing multipart streams.
@@ -39,8 +38,6 @@ public class MultipartInputStream extends ExtendedInputStream
{
// List of PartInputStream, one for each part of the key
private final List<? extends PartInputStream> partStreams;
- private final boolean isStreamBlockInputStream;
- private final boolean statelessPositionedReadSupported;
// partOffsets[i] stores the index of the first data byte in
// partStream w.r.t the whole key data.
@@ -50,7 +47,7 @@ public class MultipartInputStream extends ExtendedInputStream
{
// part[0]), partOffsets[1] = 200 and so on.
private final long[] partOffsets;
- private boolean closed;
+ private volatile boolean closed;
// Index of the partStream corresponding to the current position of the
// MultipartCryptoKeyInputStream.
private int partIndex;
@@ -67,30 +64,18 @@ public MultipartInputStream(String keyName,
this.key = keyName;
this.partStreams = Collections.unmodifiableList(inputStreams);
- this.isStreamBlockInputStream = !inputStreams.isEmpty() &&
inputStreams.get(0) instanceof StreamBlockInputStream;
// Calculate and update the partOffsets
this.partOffsets = new long[inputStreams.size()];
- boolean statelessSupported = !inputStreams.isEmpty();
int i = 0;
long streamLength = 0L;
for (PartInputStream partInputStream : inputStreams) {
this.partOffsets[i++] = streamLength;
- if (isStreamBlockInputStream) {
- Preconditions.assertInstanceOf(partInputStream,
StreamBlockInputStream.class);
- } else if (statelessSupported && !(partInputStream instanceof
BlockInputStream)) {
- statelessSupported = false;
- }
streamLength += partInputStream.getLength();
}
- this.statelessPositionedReadSupported = statelessSupported;
this.length = streamLength;
}
- public boolean isStreamBlockInputStream() {
- return isStreamBlockInputStream;
- }
-
@Override
protected synchronized int readWithStrategy(ByteReaderStrategy strategy)
throws IOException {
@@ -195,45 +180,28 @@ public synchronized void seek(long pos) throws
IOException {
}
@Override
- public boolean readFully(long position, ByteBuffer buffer) throws
IOException {
- if (!statelessPositionedReadSupported) {
- return false;
- }
- return readFullyStateless(position, buffer);
- }
-
- /**
- * Shared stateless positioned-read loop across parts using {@link
PartInputStream#readPositioned}.
- */
- private boolean readFullyStateless(long position, ByteBuffer buffer) throws
IOException {
+ protected int readPositioned(long position, ByteBuffer buffer) throws
IOException {
+ checkOpen();
if (!buffer.hasRemaining()) {
- return true;
+ return 0;
}
-
long pos = position;
int bytesRead = 0;
- while (buffer.hasRemaining()) {
- if (pos < 0 || pos >= length) {
- if (bytesRead > 0) {
- return true;
- }
- throw new EOFException("EOF encountered at pos: " + pos +
- " for key: " + key);
+ while (buffer.hasRemaining() && pos < length) {
+ int index = binarySearchOffsetIndex(partOffsets, pos);
+ PartInputStream part = partStreams.get(index);
+ long partPosition = pos - partOffsets[index];
+ final int n = part.read(partPosition, buffer);
+ if (n < 0) {
+ break;
}
- int idx = binarySearchOffsetIndex(partOffsets, pos);
- long partPos = pos - partOffsets[idx];
- int n = partStreams.get(idx).readPositioned(partPos, buffer);
- if (n <= 0) {
- if (bytesRead > 0) {
- return true;
- }
- throw new EOFException("EOF encountered at pos: " + pos +
- " for key: " + key);
+ if (n == 0) {
+ throw new IOException("No progress reading key " + key + " at position
" + pos);
}
bytesRead += n;
pos += n;
}
- return true;
+ return bytesRead == 0 ? EOF : bytesRead;
}
static int binarySearchOffsetIndex(long[] offsets, long pos) {
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 3d7ee44c58e..18f89053451 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,15 +18,16 @@
package org.apache.hadoop.hdds.scm.storage;
import java.io.IOException;
-import java.nio.ByteBuffer;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
import org.apache.hadoop.fs.CanUnbuffer;
import org.apache.hadoop.fs.Seekable;
/**
* A stream that can be a part of a {@link MultipartInputStream}.
+ * Positioned reads must allow concurrent calls without changing the
sequential cursor.
*/
public interface PartInputStream
- extends CanUnbuffer, Seekable {
+ extends CanUnbuffer, Seekable, ByteBufferPositionedReadable {
long getLength();
default long getRemaining() throws IOException {
@@ -34,16 +35,4 @@ 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 8d936215b66..c162553ee81 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
@@ -185,12 +185,12 @@ synchronized int readFully(ByteBuffer targetBuf, boolean
preRead) throws IOExcep
*
* @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.
+ * @return the number of bytes copied into {@code dst}, or -1 at EOF.
*/
@Override
public int readPositioned(long blockOffset, ByteBuffer dst) throws
IOException {
if (!dst.hasRemaining()) {
- return EOF;
+ return 0;
}
if (blockOffset < 0 || blockOffset >= blockLength) {
return EOF;
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java
index eb876642bef..b655e5d3a39 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java
@@ -31,6 +31,7 @@
import java.util.TreeMap;
import java.util.function.Function;
import org.apache.hadoop.fs.FSExceptionMessages;
+import org.apache.hadoop.fs.StreamCapabilities;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.client.ContainerBlockID;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
@@ -300,6 +301,17 @@ protected long remaining() {
return blockLength() - position;
}
+ @Override
+ protected int readPositioned(long offset, ByteBuffer buffer) {
+ throw new UnsupportedOperationException("Use ECBlockInputStreamProxy for
positioned reads");
+ }
+
+ @Override
+ public boolean hasCapability(String capability) {
+ return !StreamCapabilities.PREADBYTEBUFFER.equalsIgnoreCase(capability)
+ && super.hasCapability(capability);
+ }
+
@Override
public synchronized int read(byte[] b, int off, int len) throws IOException {
return read(ByteBuffer.wrap(b, off, len));
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java
index 3411f028841..c9d71f128b5 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java
@@ -19,10 +19,10 @@
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Function;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
@@ -56,8 +56,8 @@ public class ECBlockInputStreamProxy extends
BlockExtendedInputStream {
private BlockExtendedInputStream blockReader;
private boolean reconstructionReader = false;
- private List<DatanodeDetails> failedLocations = new ArrayList<>();
- private boolean closed = false;
+ private final CopyOnWriteArrayList<DatanodeDetails> failedLocations;
+ private volatile boolean closed = false;
private OzoneClientConfig config;
/**
@@ -102,12 +102,21 @@ public ECBlockInputStreamProxy(ECReplicationConfig
repConfig,
BlockLocationInfo> refreshFunction,
ECBlockInputStreamFactory streamFactory,
OzoneClientConfig config) {
+ this(repConfig, blockInfo, xceiverClientFactory, refreshFunction,
streamFactory, config,
+ new CopyOnWriteArrayList<>());
+ }
+
+ private ECBlockInputStreamProxy(ECReplicationConfig repConfig,
BlockLocationInfo blockInfo,
+ XceiverClientFactory xceiverClientFactory, Function<BlockID,
BlockLocationInfo> refreshFunction,
+ ECBlockInputStreamFactory streamFactory, OzoneClientConfig config,
+ CopyOnWriteArrayList<DatanodeDetails> failedLocations) {
this.repConfig = repConfig;
this.blockInfo = blockInfo;
this.ecBlockInputStreamFactory = streamFactory;
this.xceiverClientFactory = xceiverClientFactory;
this.refreshFunction = refreshFunction;
this.config = config;
+ this.failedLocations = failedLocations;
setReaderType();
createBlockReader();
@@ -116,9 +125,10 @@ public ECBlockInputStreamProxy(ECReplicationConfig
repConfig,
private synchronized void setReaderType() {
int expected = expectedDataLocations(repConfig, getLength());
int available = availableDataLocations(blockInfo.getPipeline(), expected);
- reconstructionReader = available < expected;
+ reconstructionReader = available < expected || !failedLocations.isEmpty();
if (reconstructionReader) {
- LOG.info("Data locations available: {} < expected: {}, using
reconstruction read", available, expected);
+ LOG.info("Data locations available: {}, expected: {}, failed: {}, using
reconstruction read",
+ available, expected, failedLocations.size());
}
}
@@ -197,11 +207,28 @@ public synchronized int read(ByteBuffer buf) throws
IOException {
return totalRead;
}
+ @Override
+ protected int readPositioned(long position, ByteBuffer buffer) throws
IOException {
+ ensureNotClosed();
+ if (!buffer.hasRemaining()) {
+ return 0;
+ }
+ if (position >= blockInfo.getLength()) {
+ return EOF;
+ }
+ // Each request owns its cursor and reconstruction buffers; failed
locations are shared.
+ try (ECBlockInputStreamProxy reader = new
ECBlockInputStreamProxy(repConfig, blockInfo,
+ xceiverClientFactory, refreshFunction, ecBlockInputStreamFactory,
config, failedLocations)) {
+ reader.seek(position);
+ return reader.read(buffer);
+ }
+ }
+
private synchronized void failoverToReconstructionRead(
List<DatanodeDetails> badLocations, long lastPosition)
throws IOException {
if (badLocations != null) {
- failedLocations.addAll(badLocations);
+ failedLocations.addAllAbsent(badLocations);
}
blockReader.close();
reconstructionReader = true;
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedInputStream.java
index 0d0c2ad438c..256d41eeac3 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedInputStream.java
@@ -20,6 +20,7 @@
import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
+import org.apache.hadoop.fs.StreamCapabilities;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
@@ -49,6 +50,17 @@ public ECBlockReconstructedInputStream(ECReplicationConfig
repConfig,
this.stripeReader = stripeReader;
}
+ @Override
+ protected int readPositioned(long offset, ByteBuffer buffer) {
+ throw new UnsupportedOperationException("Use ECBlockInputStreamProxy for
positioned reads");
+ }
+
+ @Override
+ public boolean hasCapability(String capability) {
+ return !StreamCapabilities.PREADBYTEBUFFER.equalsIgnoreCase(capability)
+ && super.hasCapability(capability);
+ }
+
@Override
public synchronized BlockID getBlockID() {
return stripeReader.getBlockID();
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/PositionedReadTestHelper.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/PositionedReadTestHelper.java
index fe5514698ff..42fef84799e 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/PositionedReadTestHelper.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/PositionedReadTestHelper.java
@@ -60,7 +60,8 @@ public static void runConcurrentPositionedReads(byte[] source,
final int threadId = t;
futures.add(pool.submit((Callable<Void>) () -> {
for (int i = 0; i < ITERATIONS; i++) {
- int offset = (threadId * 1000 + i * 17) % (source.length -
BUFFER_SIZE);
+ int offset = (int) (((long) threadId * source.length /
THREAD_COUNT + i * 17L)
+ % (source.length - BUFFER_SIZE));
ByteBuffer buf = ByteBuffer.allocate(BUFFER_SIZE);
action.readAtOffset(offset, buf);
buf.flip();
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java
index baf4d7fef09..22416df3af4 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java
@@ -538,6 +538,17 @@ private static Stream<Arguments>
exceptionsTriggersRefresh() {
);
}
+ @Test
+ public void testPositionedReadAtEof() throws Exception {
+ ByteBuffer empty = ByteBuffer.allocate(0);
+ for (long position : new long[] {0, blockSize, blockSize + 1}) {
+ assertEquals(0, blockStream.readPositioned(position, empty));
+ assertEquals(0, blockStream.read(position, empty));
+ }
+ assertEquals(-1, blockStream.readPositioned(blockSize,
ByteBuffer.allocate(1)));
+ assertEquals(-1, blockStream.readPositioned(blockSize + 1,
ByteBuffer.allocate(1)));
+ }
+
@Test
public void testConcurrentPositionedRead() throws Exception {
try (BlockInputStream largeBlockStream = createLargeBlockStream()) {
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java
index 96d911319f0..457d7a0a0ec 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java
@@ -189,6 +189,9 @@ public void testPositionedReadAtEof() throws Exception {
ByteBuffer dst = ByteBuffer.allocate(10);
assertEquals(-1, chunkStream.readPositioned(CHUNK_SIZE, dst));
assertEquals(-1, chunkStream.readPositioned(CHUNK_SIZE + 1, dst));
+ for (long position : new long[] {0, CHUNK_SIZE, CHUNK_SIZE + 1}) {
+ assertEquals(0, chunkStream.readPositioned(position,
ByteBuffer.allocate(0)));
+ }
}
@Test
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 7ac64376390..8eb81ed7069 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
@@ -19,8 +19,9 @@
import static
org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE;
import static
org.apache.hadoop.hdds.scm.storage.TestChunkInputStream.generateRandomData;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
@@ -30,8 +31,10 @@
import static org.mockito.Mockito.when;
import com.google.common.primitives.Bytes;
+import java.io.EOFException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -47,6 +50,8 @@
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.ozone.common.Checksum;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.mockito.invocation.InvocationOnMock;
/**
@@ -75,8 +80,15 @@ public void testPositionedReadPartialAtEof() throws
Exception {
int position = 12;
int expectedBytes = fileLen - position;
ByteBuffer buffer = ByteBuffer.allocate(expectedBytes * 2);
- assertTrue(multipartStream.readFully(position, buffer));
+ assertEquals(expectedBytes, multipartStream.read(position, buffer));
+ buffer.clear();
+ assertThrows(EOFException.class, () ->
multipartStream.readFully(position, buffer));
assertEquals(expectedBytes, buffer.position());
+ ByteBuffer empty = ByteBuffer.allocate(0);
+ for (long offset : new long[] {0, fileLen, fileLen + 1}) {
+ assertEquals(0, multipartStream.readPositioned(offset, empty));
+ assertEquals(0, multipartStream.read(offset, empty));
+ }
}
}
@@ -88,11 +100,12 @@ public void testStreamBlockPositionedReadPartialAtEof()
throws Exception {
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, multipartStream.read(position, buffer));
+ buffer.clear();
+ assertThrows(EOFException.class, () ->
multipartStream.readFully(position, buffer));
assertEquals(expectedBytes, buffer.position());
verify(part, never()).seek(anyLong());
}
@@ -111,20 +124,16 @@ public void testStreamBlockConcurrentPositionedRead()
throws Exception {
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);
- }
- });
+ multipartStream::readFully);
verify(part0, never()).seek(anyLong());
verify(part1, never()).seek(anyLong());
}
}
- @Test
- public void testConcurrentPositionedRead() throws Exception {
+ @ParameterizedTest
+ @CsvSource({"false, false", "true, false", "false, true"})
+ public void testConcurrentPositionedRead(boolean streamPart0, boolean
streamPart1) throws Exception {
byte[] part0Data = generateRandomData(PART_SIZE);
byte[] part1Data = generateRandomData(PART_SIZE);
byte[] keyData = Bytes.concat(part0Data, part1Data);
@@ -136,22 +145,23 @@ public void testConcurrentPositionedRead() throws
Exception {
Function<BlockID, BlockLocationInfo> refreshFunction =
mock(Function.class);
Checksum checksum = new Checksum(ChecksumType.NONE, 1024);
- BlockInputStream part0 = createBlockStream(new BlockID(new
ContainerBlockID(1, 1)),
+ PartInputStream part0 = streamPart0 ? stubStreamBlockPart(part0Data)
+ : createBlockStream(new BlockID(new ContainerBlockID(1, 1)),
part0Data, pipeline, refreshFunction, clientConfig, checksum);
- BlockInputStream part1 = createBlockStream(new BlockID(new
ContainerBlockID(1, 2)),
+ PartInputStream part1 = streamPart1 ? stubStreamBlockPart(part1Data)
+ : createBlockStream(new BlockID(new ContainerBlockID(1, 2)),
part1Data, pipeline, refreshFunction, clientConfig, checksum);
- List<BlockInputStream> parts = new ArrayList<>();
+ List<PartInputStream> parts = new ArrayList<>();
parts.add(part0);
parts.add(part1);
try (MultipartInputStream multipartStream = new
MultipartInputStream("test-key", parts)) {
- multipartStream.initialize();
PositionedReadTestHelper.runConcurrentPositionedReads(keyData,
- (offset, buf) -> {
- if (!multipartStream.readFully(offset, buf)) {
- throw new AssertionError("stateless readFully returned false at
" + offset);
- }
- });
+ multipartStream::readFully);
+ ByteBuffer buffer = ByteBuffer.allocate(4);
+ multipartStream.readFully(PART_SIZE - 2, buffer);
+ assertArrayEquals(Arrays.copyOfRange(keyData, PART_SIZE - 2, PART_SIZE +
2), buffer.array());
+ assertEquals(0, multipartStream.getPos());
}
}
@@ -159,7 +169,7 @@ private static StreamBlockInputStream
stubStreamBlockPart(byte[] partData) throw
StreamBlockInputStream part = mock(StreamBlockInputStream.class);
when(part.getLength()).thenReturn((long) partData.length);
doAnswer(invocation -> readPositionedFrom(partData, invocation)).when(part)
- .readPositioned(anyLong(), any(ByteBuffer.class));
+ .read(anyLong(), any(ByteBuffer.class));
return part;
}
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 6132f1e9dc5..ceb4497fd33 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
@@ -24,6 +24,7 @@
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.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.eq;
@@ -36,6 +37,7 @@
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.Duration;
@@ -45,6 +47,11 @@
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
@@ -65,6 +72,7 @@
import org.apache.hadoop.hdds.scm.StreamingReaderSpi;
import org.apache.hadoop.hdds.scm.XceiverClientFactory;
import org.apache.hadoop.hdds.scm.XceiverClientGrpc;
+import org.apache.hadoop.hdds.scm.XceiverClientSpi;
import
org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
@@ -87,6 +95,95 @@ public class TestStreamBlockInputStream {
private static final Duration STREAM_READ_TIMEOUT = Duration.ofSeconds(5);
private static final Function<BlockID, BlockLocationInfo> NO_REFRESH = b ->
null;
+ @Test
+ public void testPositionedReadsOverlapWithoutMovingCursor() throws Exception
{
+ XceiverClientGrpc client = mock(XceiverClientGrpc.class);
+ XceiverClientFactory factory = mock(XceiverClientFactory.class);
+ when(factory.acquireClientForReadData(any())).thenReturn(client);
+ Map<StreamingReadResponse, StreamingReaderSpi> readers = new
ConcurrentHashMap<>();
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = inv.getArgument(1);
+ StreamingReadResponse response = new StreamingReadResponse(
+ MockDatanodeDetails.randomDatanodeDetails(),
mock(ClientCallStreamObserver.class));
+ reader.setStreamingReadResponse(response);
+ readers.put(response, reader);
+ return null;
+ }).when(client).initStreamRead(any(), any(), any());
+ CountDownLatch started = new CountDownLatch(2);
+ doAnswer(inv -> {
+ ContainerCommandRequestProto request = inv.getArgument(0);
+ started.countDown();
+ assertTrue(started.await(5, TimeUnit.SECONDS), "positioned reads must
overlap");
+ long offset = request.getReadBlock().getOffset();
+ assertEquals(2, request.getReadBlock().getLength(), "positioned reads
must not request read-ahead");
+ StreamingReaderSpi reader = readers.get(inv.getArgument(1));
+ reader.onNext(buildResponseProto(new byte[] {(byte) offset}, offset));
+ reader.onNext(buildResponseProto(new byte[] {(byte) (offset + 1)},
offset + 1));
+ reader.onCompleted();
+ return null;
+ }).when(client).streamRead(any(), any());
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try (StreamBlockInputStream stream = new StreamBlockInputStream(new
BlockID(1, 22), 100,
+ mockStandalonePipeline(), null, factory, NO_REFRESH,
newStreamReadConfig())) {
+ stream.seek(50);
+ Future<?> first = executor.submit(() -> {
+ ByteBuffer buffer = ByteBuffer.allocate(2);
+ stream.readFully(10, buffer);
+ assertArrayEquals(new byte[] {10, 11}, buffer.array());
+ return null;
+ });
+ Future<?> second = executor.submit(() -> {
+ byte[] buffer = new byte[2];
+ stream.readFully(20, buffer);
+ assertArrayEquals(new byte[] {20, 21}, buffer);
+ return null;
+ });
+ first.get(10, TimeUnit.SECONDS);
+ second.get(10, TimeUnit.SECONDS);
+ assertEquals(50, stream.getPos());
+ } finally {
+ executor.shutdownNow();
+ }
+ verify(factory, times(2)).releaseClientForReadData(client, false);
+ verify(client, times(2)).completeStreamRead();
+ for (StreamingReadResponse response : readers.keySet()) {
+ verify(response.getRequestObserver()).onCompleted();
+ }
+ }
+
+ @Test
+ public void testIncompletePositionedReadReleasesResources() throws Exception
{
+ ClientCallStreamObserver<ContainerCommandRequestProto> observer =
mock(ClientCallStreamObserver.class);
+ XceiverClientGrpc client = mockCapturingStreamingReadClient(observer,
StreamingReaderSpi::onCompleted);
+ XceiverClientFactory factory = mock(XceiverClientFactory.class);
+ when(factory.acquireClientForReadData(any())).thenReturn(client);
+ try (StreamBlockInputStream stream = new StreamBlockInputStream(new
BlockID(1, 23), 100,
+ mockStandalonePipeline(), null, factory, NO_REFRESH,
newStreamReadConfig())) {
+ stream.seek(50);
+ ByteBuffer buffer = ByteBuffer.allocate(2);
+ assertThrows(EOFException.class, () -> stream.readFully(10, buffer));
+ assertEquals(0, buffer.position());
+ assertEquals(50, stream.getPos());
+ }
+ verify(factory).releaseClientForReadData(client, false);
+ verify(client).completeStreamRead();
+ verify(observer).onCompleted();
+ }
+
+ @Test
+ public void testFailedPositionedReadInitializationReleasesClient() throws
Exception {
+ XceiverClientGrpc client = mock(XceiverClientGrpc.class);
+ doThrow(new IOException("initialization
failed")).when(client).initStreamRead(any(), any(), any());
+ XceiverClientFactory factory = mock(XceiverClientFactory.class);
+ when(factory.acquireClientForReadData(any())).thenReturn(client);
+ try (StreamBlockInputStream stream = new StreamBlockInputStream(new
BlockID(1, 24), 100,
+ mockStandalonePipeline(), null, factory, NO_REFRESH,
newStreamReadConfig())) {
+ assertThrows(IOException.class, () -> stream.readFully(10,
ByteBuffer.allocate(2)));
+ }
+ verify(factory).releaseClientForReadData(client, false);
+ verify(client, never()).completeStreamRead();
+ }
+
@Test
public void testCustomStreamReadConfigIsApplied() throws Exception {
// Arrange: create a config with non-default values
@@ -152,8 +249,12 @@ public void testPositionedReadAtEof() throws Exception {
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)));
+ ByteBuffer empty = ByteBuffer.allocate(1);
+ empty.position(1);
+ for (long position : new long[] {0, data.length, data.length + 1}) {
+ assertEquals(0, sbis.readPositioned(position, empty));
+ assertEquals(0, sbis.read(position, empty));
+ }
}
}
@@ -183,6 +284,18 @@ public void
testPositionedReadDoesNotReleasePermitWhenInitStreamReadFails() thro
verify(xceiverClient, never()).completeStreamRead();
}
+ @Test
+ public void testPositionedReadReleasesUnexpectedClient() throws Exception {
+ XceiverClientSpi client = mock(XceiverClientSpi.class);
+ XceiverClientFactory factory = mock(XceiverClientFactory.class);
+
when(factory.acquireClientForReadData(any(Pipeline.class))).thenReturn(client);
+ try (StreamBlockInputStream stream = new StreamBlockInputStream(
+ new BlockID(1, 25), 64, mockStandalonePipeline(), null, factory,
NO_REFRESH, newStreamReadConfig())) {
+ assertThrows(IOException.class, () -> stream.readPositioned(0,
ByteBuffer.allocate(1)));
+ }
+ verify(factory).releaseClientForReadData(client, false);
+ }
+
@Test
public void testConcurrentPositionedRead() throws Exception {
byte[] data = TestChunkInputStream.generateRandomData(SOURCE_SIZE);
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/ECStreamTestUtil.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/ECStreamTestUtil.java
index 639b4e92a14..d84540ec8ac 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/ECStreamTestUtil.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/ECStreamTestUtil.java
@@ -266,7 +266,7 @@ public synchronized BlockExtendedInputStream create(
int repInd = currentPipeline.getReplicaIndex(pipeline.getNodes().get(0));
TestBlockInputStream stream = new TestBlockInputStream(
blockInfo.getBlockID(), blockInfo.getLength(),
- blockStreamData.get(repInd - 1), repInd);
+ blockStreamData.get(repInd - 1).duplicate(), repInd);
if (failIndexes.remove(Integer.valueOf(repInd))) {
stream.setShouldError(true);
}
@@ -324,6 +324,19 @@ public int getEcReplicaIndex() {
return ecReplicaIndex;
}
+ @Override
+ protected int readPositioned(long position, ByteBuffer buffer) {
+ if (position >= length) {
+ return EOF;
+ }
+ ByteBuffer source = data.duplicate();
+ source.position((int) position);
+ int n = (int) Math.min(buffer.remaining(), length - position);
+ source.limit(source.position() + n);
+ buffer.put(source);
+ return n;
+ }
+
@Override
public BlockID getBlockID() {
return blockID;
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockInputStream.java
index 97f9230b471..e85ca219b86 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockInputStream.java
@@ -659,6 +659,18 @@ public void setSimulateZeroByteRead(boolean
simulateZeroByteRead) {
this.simulateZeroByteRead = simulateZeroByteRead;
}
+ @Override
+ protected int readPositioned(long offset, ByteBuffer buffer) {
+ if (offset >= length) {
+ return EOF;
+ }
+ int n = (int) Math.min(buffer.remaining(), length - offset);
+ for (int i = 0; i < n; i++) {
+ buffer.put(dataVal);
+ }
+ return n;
+ }
+
@Override
public BlockID getBlockID() {
return blockID;
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockInputStreamProxy.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockInputStreamProxy.java
index 074e6f1f9a8..42162960c2f 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockInputStreamProxy.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockInputStreamProxy.java
@@ -18,16 +18,26 @@
package org.apache.hadoop.ozone.client.io;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.SplittableRandom;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
@@ -40,6 +50,8 @@
import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
/**
* Unit tests for the ECBlockInputStreamProxy class.
@@ -328,6 +340,131 @@ public void testCanSeekToNewPosition() throws IOException
{
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void positionedReadsRememberFailedLocations(boolean sequentialFirst) throws
Exception {
+ byte[] data = generateData(1024).array();
+ BlockLocationInfo info = ECStreamTestUtil.createKeyInfo(repConfig,
data.length,
+ ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5));
+ DatanodeDetails bad = info.getPipeline().getFirstNode();
+ AtomicInteger failedAttempts = new AtomicInteger();
+ ECBlockInputStreamFactory factory = (missing, failed, replication, block,
xceiver, refresh, config) -> {
+ if (missing) {
+ assertThat(failed).containsExactly(bad);
+ }
+ return new ECStreamTestUtil.TestBlockInputStream(block.getBlockID(),
block.getLength(), ByteBuffer.wrap(data)) {
+ @Override
+ public int read(ByteBuffer destination) throws IOException {
+ if (!missing) {
+ failedAttempts.incrementAndGet();
+ throw new BadDataLocationException(bad, "failed data replica");
+ }
+ return super.read(destination);
+ }
+ };
+ };
+ try (ECBlockInputStreamProxy stream = new
ECBlockInputStreamProxy(repConfig, info, null, null,
+ factory, conf.getObject(OzoneClientConfig.class))) {
+ ByteBuffer first = ByteBuffer.allocate(10);
+ assertEquals(10, sequentialFirst ? stream.read(first) : stream.read(0,
first));
+ for (int offset : new int[] {17, 203}) {
+ ByteBuffer buffer = ByteBuffer.allocate(20);
+ stream.readFully(offset, buffer);
+ assertArrayEquals(Arrays.copyOfRange(data, offset, offset + 20),
buffer.array());
+ }
+ assertEquals(1, failedAttempts.get());
+ assertEquals(sequentialFirst ? 10 : 0, stream.getPos());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void emptyPositionedReadsDoNotCreateReaders(boolean missing) throws
Exception {
+ int length = 100;
+ generateData(length);
+ BlockLocationInfo info = ECStreamTestUtil.createKeyInfo(repConfig, length,
+ missing ? ECStreamTestUtil.createIndexMap(2, 3, 4, 5) :
ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5));
+ try (ECBlockInputStreamProxy stream = createBISProxy(repConfig, info)) {
+ BlockExtendedInputStream reader =
streamFactory.getStreams().get(missing);
+ ByteBuffer empty = ByteBuffer.allocate(0);
+ for (long position : new long[] {0, length, length + 1}) {
+ assertEquals(0, stream.readPositioned(position, empty));
+ assertEquals(0, stream.read(position, empty));
+ }
+ assertThat(streamFactory.getStreams().get(missing)).isSameAs(reader);
+ assertEquals(-1, stream.readPositioned(length, ByteBuffer.allocate(1)));
+ assertEquals(-1, stream.readPositioned(length + 1,
ByteBuffer.allocate(1)));
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"healthy", "missing", "failover", "failure"})
+ void positionedReadsOverlapAndCloseTheirReaders(String mode) throws
Exception {
+ byte[] data = generateData(1024).array();
+ BlockLocationInfo info = ECStreamTestUtil.createKeyInfo(repConfig,
data.length,
+ "missing".equals(mode) ? ECStreamTestUtil.createIndexMap(2, 3, 4, 5)
+ : ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5));
+ CyclicBarrier reads = new CyclicBarrier(2);
+ AtomicInteger opened = new AtomicInteger();
+ AtomicInteger closed = new AtomicInteger();
+ ECBlockInputStreamFactory factory = (missing, failed, replication, block,
xceiver, refresh, config) -> {
+ opened.incrementAndGet();
+ return new ECStreamTestUtil.TestBlockInputStream(block.getBlockID(),
block.getLength(), ByteBuffer.wrap(data)) {
+ @Override
+ public int read(ByteBuffer destination) throws IOException {
+ if ("failover".equals(mode) && !missing) {
+ destination.put((byte) 0); // Retry must overwrite bytes from the
failed attempt.
+ throw new
BadDataLocationException(info.getPipeline().getFirstNode(), "failed data
replica");
+ }
+ try {
+ reads.await(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new IOException("EC positioned reads did not overlap", e);
+ }
+ if ("failure".equals(mode)) {
+ throw new IOException("injected failure");
+ }
+ return super.read(destination);
+ }
+
+ @Override
+ public void close() {
+ closed.incrementAndGet();
+ }
+ };
+ };
+ ExecutorService pool = Executors.newFixedThreadPool(2);
+ try (ECBlockInputStreamProxy stream = new
ECBlockInputStreamProxy(repConfig, info, null, null,
+ factory, conf.getObject(OzoneClientConfig.class))) {
+ stream.seek(71);
+ List<Future<?>> futures = new ArrayList<>();
+ for (int offset : new int[] {17, 203}) {
+ futures.add(pool.submit((Callable<Void>) () -> {
+ ByteBuffer destination = ByteBuffer.allocateDirect(333);
+ if ("failure".equals(mode)) {
+ assertThrows(IOException.class, () -> stream.readFully(offset,
destination));
+ } else {
+ stream.readFully(offset, destination);
+ destination.flip();
+ byte[] actual = new byte[destination.remaining()];
+ destination.get(actual);
+ assertArrayEquals(Arrays.copyOfRange(data, offset, offset + 333),
actual);
+ }
+ return null;
+ }));
+ }
+ for (Future<?> future : futures) {
+ future.get(20, TimeUnit.SECONDS);
+ }
+ assertEquals(71, stream.getPos());
+ assertEquals(opened.get() - 1, closed.get());
+ assertEquals(-1, stream.read(data.length, ByteBuffer.allocate(1)));
+ } finally {
+ pool.shutdownNow();
+ }
+ assertEquals(opened.get(), closed.get());
+ }
+
private ByteBuffer generateData(int length) {
ByteBuffer data = ByteBuffer.allocate(length);
ECStreamTestUtil.randomFill(data, dataGenerator);
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockReconstructedInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockReconstructedInputStream.java
index 91914494f14..27a01f21eee 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockReconstructedInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestECBlockReconstructedInputStream.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.ozone.client.io;
import static
org.apache.hadoop.ozone.client.io.ECStreamTestUtil.generateParity;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -28,9 +29,12 @@
import java.util.List;
import java.util.Map;
import java.util.SplittableRandom;
+import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -363,6 +367,41 @@ private ByteBuffer[] allocateBuffers(int num, int size) {
return bufs;
}
+ @Test
+ void concurrentPositionedReadsReconstructIndependentStripes() throws
Exception {
+ int chunk = repConfig.getEcChunkSize();
+ int blockLength = chunk * 7 - 101;
+ ByteBuffer[] data = allocateBuffers(repConfig.getData(), chunk * 3);
+ ECStreamTestUtil.randomFill(data, chunk, dataGenerator, blockLength);
+ addDataStreamsToFactory(data, generateParity(data, repConfig));
+ byte[] expected = new byte[blockLength];
+ ECStreamTestUtil.randomFill(ByteBuffer.wrap(expected), new
SplittableRandom(randomSeed));
+ BlockLocationInfo info = ECStreamTestUtil.createKeyInfo(repConfig,
blockLength,
+ ECStreamTestUtil.createIndexMap(2, 3, 4, 5));
+ streamFactory.setCurrentPipeline(info.getPipeline());
+ ECBlockInputStreamFactory factory =
ECBlockInputStreamFactoryImpl.getInstance(streamFactory, bufferPool,
+ () -> ecReconstructExecutor);
+ ExecutorService readers = Executors.newFixedThreadPool(2);
+ try (ECBlockInputStreamProxy stream = new
ECBlockInputStreamProxy(repConfig, info, null, null, factory,
+ conf.getObject(OzoneClientConfig.class))) {
+ List<Future<?>> futures = new ArrayList<>();
+ for (int offset : new int[] {chunk - 37, chunk * 3 - 29}) {
+ futures.add(readers.submit((Callable<Void>) () -> {
+ byte[] actual = new byte[chunk * 3 + 53];
+ stream.readFully(offset, actual);
+ assertArrayEquals(Arrays.copyOfRange(expected, offset, offset +
actual.length), actual);
+ return null;
+ }));
+ }
+ for (Future<?> future : futures) {
+ future.get(20, TimeUnit.SECONDS);
+ }
+ assertEquals(0, stream.getPos());
+ } finally {
+ readers.shutdownNow();
+ }
+ }
+
private void addDataStreamsToFactory(ByteBuffer[] data, ByteBuffer[] parity)
{
List<ByteBuffer> dataStreams = new ArrayList<>(Arrays.asList(data));
dataStreams.addAll(Arrays.asList(parity));
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 705aa84d637..2d2c3943ecb 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
@@ -79,17 +79,6 @@ public int getBufferSize() {
return bufferSize;
}
- /**
- * {@link CryptoInputStream} does not synchronize its own methods, so every
method moving the cursor of
- * this stream is serialized here on the monitor of this stream. Otherwise a
read or a seek could land in
- * the middle of a positioned read and see (or undo) the cursor move that
read does.
- * This covers both the sequential-read API ({@link #read()}, {@link
#read(byte[], int, int)},
- * {@link #read(ByteBuffer)}, {@link #seek(long)}, {@link #getPos()}, {@link
#skip(long)}) and all
- * positioned-read overloads
- * ({@link #read(long, ByteBuffer)}, {@link #readFully(long, ByteBuffer)},
- * {@link #read(long, byte[], int, int)}, {@link #readFully(long, byte[],
int, int)},
- * {@link #readFully(long, byte[])}).
- */
@Override
public synchronized int read(byte[] b, int off, int len) throws IOException {
// CryptoInputStream reads hadoop.security.crypto.buffer.size number of
@@ -188,91 +177,49 @@ 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:
- * the cursor is moved to {@code position}, data is read via {@link
#read(byte[], int, int)} (which
- * handles the Crypto boundary adjustment), and the cursor is restored
before the lock is released.
- *
- * <p>Returns -1 for {@code position >= length}. A read-only {@code dst} is
rejected before any state
- * is mutated. If any exception escapes the read loop the crypto-boundary
adjustment fields are reset so
- * the next sequential read does not fail the precondition in {@code
getNumBytesToRead}.
+ * Reads and decrypts at the given offset without moving the sequential
cursor.
+ * This method needs no synchronization: CryptoInputStream uses separate
buffers and a decryptor for each call.
*/
@Override
- public synchronized int read(long position, ByteBuffer dst) throws
IOException {
- if (!dst.hasRemaining()) {
- return 0;
+ public int read(long position, ByteBuffer dst) throws IOException {
+ if (position < 0) {
+ throw new EOFException("The given position is negative: " + position);
}
if (dst.isReadOnly()) {
throw new ReadOnlyBufferException();
}
- if (position < 0) {
- throw new EOFException("The given position is negative: " + position);
+ if (!dst.hasRemaining()) {
+ return 0;
}
- if (position >= getLength()) {
+ if (position >= length) {
return EOF;
}
-
- final long oldPos = getPos();
- final int readLength = (int) Math.min(dst.remaining(), getLength() -
position);
- final byte[] buffer = new byte[Math.min(readLength, getBufferSize())];
- Throwable failure = null;
+ int limit = dst.limit();
try {
- seek(position);
- int totalReadLen = 0;
- while (totalReadLen < readLength) {
- final int numBytesRead = read(buffer, 0,
- Math.min(buffer.length, readLength - totalReadLen));
- if (numBytesRead <= 0) {
- break;
- }
- dst.put(buffer, 0, numBytesRead);
- totalReadLen += numBytesRead;
- }
- return totalReadLen == 0 ? EOF : totalReadLen;
- } catch (Throwable t) {
- failure = t;
- // Defensively reset crypto-boundary adjustment fields so that the next
- // sequential read does not fail the precondition check in
getNumBytesToRead.
- readPositionAdjustedBy = 0;
- readLengthAdjustedBy = 0;
- throw t;
+ dst.limit(dst.position() + (int) Math.min(dst.remaining(), length -
position));
+ return super.read(position, dst);
} finally {
- try {
- seek(oldPos);
- } catch (IOException e) {
- if (failure == null) {
- throw e;
- }
- failure.addSuppressed(e);
- }
+ dst.limit(limit);
}
}
@Override
- public synchronized void readFully(long position, ByteBuffer dst) throws
IOException {
+ public void readFully(long position, ByteBuffer dst) throws IOException {
int bytesRead;
for (int readCount = 0; dst.hasRemaining(); readCount += bytesRead) {
bytesRead = read(position + readCount, dst);
if (bytesRead < 0) {
throw new EOFException(FSExceptionMessages.EOF_IN_READ_FULLY);
}
+ if (bytesRead == 0) {
+ throw new IOException("No progress reading at position " + (position +
readCount));
+ }
}
}
- /**
- * Byte-array positioned read. Routes through {@link #read(long,
ByteBuffer)} so that the
- * seek-read-restore sequence is covered by the same monitor as all other
cursor-moving operations.
- * Guards against null buffer (IAE) and negative position (EOFException)
before touching ByteBuffer,
- * aligning with the {@link org.apache.hadoop.fs.PositionedReadable}
contract.
- */
@Override
- public synchronized int read(long position, byte[] buffer, int offset, int
len) throws IOException {
+ public int read(long position, byte[] buffer, int offset, int len) throws
IOException {
if (buffer == null) {
throw new IllegalArgumentException("Null buffer");
}
@@ -283,7 +230,7 @@ public synchronized int read(long position, byte[] buffer,
int offset, int len)
}
@Override
- public synchronized void readFully(long position, byte[] buffer, int offset,
int len) throws IOException {
+ public void readFully(long position, byte[] buffer, int offset, int len)
throws IOException {
if (buffer == null) {
throw new IllegalArgumentException("Null buffer");
}
@@ -294,7 +241,7 @@ public synchronized void readFully(long position, byte[]
buffer, int offset, int
}
@Override
- public synchronized void readFully(long position, byte[] buffer) throws
IOException {
+ public void readFully(long position, byte[] buffer) throws IOException {
readFully(position, buffer, 0, buffer.length);
}
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java
index 2366893ef56..2d3e93deb31 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java
@@ -20,16 +20,19 @@
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
import org.apache.hadoop.fs.ByteBufferReadable;
import org.apache.hadoop.fs.CanUnbuffer;
import org.apache.hadoop.fs.Seekable;
+import org.apache.hadoop.fs.StreamCapabilities;
+import org.apache.hadoop.hdds.scm.storage.ByteReaderStrategy;
+import org.apache.hadoop.hdds.scm.storage.ExtendedInputStream;
/**
* OzoneInputStream is used to read data from Ozone.
* It uses {@link KeyInputStream} for reading the data.
*/
-public class OzoneInputStream extends InputStream implements CanUnbuffer,
- ByteBufferReadable, Seekable {
+public class OzoneInputStream extends ExtendedInputStream {
private final InputStream inputStream;
@@ -66,6 +69,38 @@ public int read(ByteBuffer byteBuffer) throws IOException {
}
}
+ @Override
+ protected int readWithStrategy(ByteReaderStrategy strategy) throws
IOException {
+ return strategy.readFromBlock(inputStream, strategy.getTargetLength());
+ }
+
+ @Override
+ protected int readPositioned(long position, ByteBuffer buffer) throws
IOException {
+ if (inputStream instanceof ByteBufferPositionedReadable) {
+ return ((ByteBufferPositionedReadable) inputStream).read(position,
buffer);
+ }
+ throw new UnsupportedOperationException("Positioned reads are not
supported by "
+ + inputStream.getClass().getName());
+ }
+
+ @Override
+ public boolean hasCapability(String capability) {
+ if (!super.hasCapability(capability)) {
+ return false;
+ }
+ if (inputStream instanceof StreamCapabilities
+ && !((StreamCapabilities) inputStream).hasCapability(capability)) {
+ return false;
+ }
+ if (StreamCapabilities.PREADBYTEBUFFER.equalsIgnoreCase(capability)) {
+ return inputStream instanceof ByteBufferPositionedReadable;
+ }
+ if (StreamCapabilities.READBYTEBUFFER.equalsIgnoreCase(capability)) {
+ return inputStream instanceof ByteBufferReadable;
+ }
+ return StreamCapabilities.UNBUFFER.equalsIgnoreCase(capability) &&
inputStream instanceof CanUnbuffer;
+ }
+
@Override
public synchronized void close() throws IOException {
inputStream.close();
diff --git
a/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestOzoneCryptoInputStream.java
b/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestOzoneCryptoInputStream.java
index d154479d8b2..41d003b3e31 100644
---
a/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestOzoneCryptoInputStream.java
+++
b/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestOzoneCryptoInputStream.java
@@ -36,6 +36,7 @@
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -44,8 +45,10 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.crypto.CryptoCodec;
import org.apache.hadoop.crypto.Encryptor;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
import org.apache.hadoop.fs.PositionedReadable;
import org.apache.hadoop.fs.Seekable;
+import org.apache.hadoop.hdds.scm.storage.MultipartInputStream;
import org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -104,8 +107,7 @@ void testPositionedReadInMiddle() throws Exception {
@Test
void testPositionedReadCrossesBufferBoundary() throws Exception {
- // The crypto buffer is 8 KB by default; read crossing that boundary
exercises
- // the position/length adjustment logic in
OzoneCryptoInputStream.read(byte[], int, int).
+ // The positioned decryptor must handle an unaligned range across crypto
buffers.
byte[] plaintext = RandomUtils.secure().randomBytes(32 * 1024);
try (OzoneCryptoInputStream s = buildStream(plaintext)) {
// Start 100 bytes before the 8 KB boundary
@@ -319,11 +321,7 @@ void testByteArrayReadFullyDoesNotMoveSequentialCursor()
throws Exception {
@Test
@Timeout(value = 30)
void testConcurrentSkipAndPositionedRead() throws Exception {
- // Verifies that skip() is serialized on the same monitor as positioned
reads.
- // Half the threads continuously seek-then-skip (cursor-moving sequential
ops).
- // The other half do positioned readFully and assert correctness. If
skip() could
- // slip between the positioned read's internal seek and read, the read
would
- // consume data from the wrong offset and assertArrayEquals would fail.
+ // Sequential cursor changes must not affect positioned decryption.
byte[] plaintext = RandomUtils.secure().randomBytes(SOURCE_SIZE);
try (OzoneCryptoInputStream s = buildStream(plaintext)) {
ExecutorService pool = Executors.newFixedThreadPool(THREAD_COUNT);
@@ -392,8 +390,7 @@ void testConcurrentPositionedReadsByteArray() throws
Exception {
@Test
@Timeout(value = 30)
void testConcurrentPositionedReadsMixedApi() throws Exception {
- // Intermix ByteBuffer and byte-array callers on the same stream to verify
- // both APIs share the same monitor and do not interleave.
+ // Intermix ByteBuffer and byte-array callers on the same stream.
byte[] plaintext = RandomUtils.secure().randomBytes(SOURCE_SIZE);
try (OzoneCryptoInputStream s = buildStream(plaintext)) {
PositionedReadTestHelper.runConcurrentPositionedReads(plaintext,
(offset, buf) -> {
@@ -408,19 +405,59 @@ void testConcurrentPositionedReadsMixedApi() throws
Exception {
}
}
+ @Test
+ void positionedReadsOverlapAcrossEncryptedParts() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(SOURCE_SIZE);
+ int split = SOURCE_SIZE / 2 + 7;
+ CyclicBarrier reads = new CyclicBarrier(2);
+ try (MultipartInputStream stream = new MultipartInputStream(KEY_NAME,
Arrays.asList(
+ buildStream(Arrays.copyOf(plaintext, split), reads),
+ buildStream(Arrays.copyOfRange(plaintext, split, plaintext.length),
reads)))) {
+ stream.seek(777);
+ ExecutorService pool = Executors.newFixedThreadPool(2);
+ try {
+ List<Future<?>> futures = new ArrayList<>();
+ for (int offset : new int[] {split - 101, split - 203}) {
+ futures.add(pool.submit((Callable<Void>) () -> {
+ ByteBuffer destination = ByteBuffer.allocateDirect(20000);
+ destination.position(3);
+ stream.readFully(offset, destination);
+ destination.flip();
+ destination.position(3);
+ assertArrayEquals(Arrays.copyOfRange(plaintext, offset, offset +
19997), toArray(destination));
+ return null;
+ }));
+ }
+ for (Future<?> future : futures) {
+ future.get(20, TimeUnit.SECONDS);
+ }
+ assertEquals(777, stream.getPos());
+ byte[] sequential = new byte[97];
+ assertEquals(sequential.length, stream.read(sequential));
+ assertArrayEquals(Arrays.copyOfRange(plaintext, 777, 874), sequential);
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+ }
+
/**
* Encrypts {@code plaintext} with the shared key/IV, wraps it in a
* {@link SeekableByteArrayInputStream}, and returns a ready-to-use
* {@link OzoneCryptoInputStream}.
*/
private static OzoneCryptoInputStream buildStream(byte[] plaintext) throws
Exception {
+ return buildStream(plaintext, null);
+ }
+
+ private static OzoneCryptoInputStream buildStream(byte[] plaintext,
CyclicBarrier reads) throws Exception {
Encryptor enc = codec.createEncryptor();
enc.init(key, iv);
ByteBuffer plain = ByteBuffer.wrap(plaintext);
ByteBuffer cipher = ByteBuffer.allocate(plaintext.length);
enc.encrypt(plain, cipher);
- SeekableByteArrayInputStream raw = new
SeekableByteArrayInputStream(cipher.array());
+ SeekableByteArrayInputStream raw = new
SeekableByteArrayInputStream(cipher.array(), reads);
LengthInputStream lin = new LengthInputStream(raw, plaintext.length);
return new OzoneCryptoInputStream(lin, codec, key, iv, KEY_NAME,
PART_INDEX);
}
@@ -436,13 +473,15 @@ private static byte[] toArray(ByteBuffer buf) {
* Used to feed ciphertext into {@link OzoneCryptoInputStream} in tests.
*/
private static final class SeekableByteArrayInputStream extends InputStream
- implements Seekable, PositionedReadable {
+ implements Seekable, PositionedReadable, ByteBufferPositionedReadable {
private final byte[] data;
+ private final CyclicBarrier reads;
private volatile int pos;
- SeekableByteArrayInputStream(byte[] data) {
+ SeekableByteArrayInputStream(byte[] data, CyclicBarrier reads) {
this.data = data;
+ this.reads = reads;
}
@Override
@@ -479,6 +518,31 @@ public boolean seekToNewSource(long targetPos) {
return false;
}
+ @Override
+ public int read(long position, ByteBuffer destination) throws IOException {
+ if (reads != null) {
+ try {
+ reads.await(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new IOException("Positioned reads did not overlap", e);
+ }
+ }
+ if (position >= data.length) {
+ return -1;
+ }
+ int n = Math.min(destination.remaining(), data.length - (int) position);
+ destination.put(data, (int) position, n);
+ return n;
+ }
+
+ @Override
+ public void readFully(long position, ByteBuffer destination) throws
IOException {
+ int n = destination.remaining();
+ if (read(position, destination) < n) {
+ throw new EOFException();
+ }
+ }
+
@Override
public int read(long position, byte[] buffer, int offset, int length)
throws IOException {
if (position >= data.length) {
diff --git a/hadoop-ozone/integration-test/pom.xml
b/hadoop-ozone/integration-test/pom.xml
index c4e3d7dad12..e0daabd27f0 100644
--- a/hadoop-ozone/integration-test/pom.xml
+++ b/hadoop-ozone/integration-test/pom.xml
@@ -236,6 +236,12 @@
<artifactId>hdds-client</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.ozone</groupId>
+ <artifactId>hdds-client</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.ozone</groupId>
<artifactId>hdds-common</artifactId>
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
index e704f289770..9207fc42831 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
@@ -32,6 +32,7 @@
import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.file.Files;
+import java.util.Arrays;
import java.util.UUID;
import java.util.stream.Stream;
import org.apache.commons.lang3.RandomStringUtils;
@@ -43,6 +44,7 @@
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.StorageType;
+import org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.ozone.DataTestUtil;
@@ -54,6 +56,7 @@
import org.apache.ozone.test.NonHATests;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@@ -113,6 +116,41 @@ void shutdown() {
closeQuietly(client, ecFs);
}
+ @Test
+ void testConcurrentECPositionedRead() throws Exception {
+ Path path = new Path("/concurrent-ec-reads");
+ byte[] expected = Arrays.copyOf(data, 7 * 1024 * 1024 + 103);
+ try (FSDataOutputStream output = ecFs.create(path)) {
+ output.write(expected);
+ }
+ try (FSDataInputStream stream = ecFs.open(path)) {
+ stream.seek(123);
+ PositionedReadTestHelper.runConcurrentPositionedReads(expected,
stream::readFully);
+ assertEquals(123, stream.getPos());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testConcurrentPositionedRead(boolean streaming) throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration(cluster().getConf());
+ conf.setBoolean("ozone.client.stream.readblock.enable", streaming);
+ try (FileSystem fs = FileSystem.newInstance(URI.create(uri), conf);
+ FSDataInputStream stream = fs.open(filePath)) {
+ stream.seek(123);
+ PositionedReadTestHelper.runConcurrentPositionedReads(data, (offset,
buffer) -> {
+ if ((offset & 1) == 0) {
+ stream.readFully(offset, buffer);
+ } else {
+ byte[] bytes = new byte[buffer.remaining()];
+ stream.readFully(offset, bytes);
+ buffer.put(bytes);
+ }
+ });
+ assertEquals(123, stream.getPos());
+ }
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testO3FSSingleByteRead(boolean isStreamEnable) throws
IOException {
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneAtRestEncryption.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneAtRestEncryption.java
index 3bf1df6b952..06314be3d2a 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneAtRestEncryption.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneAtRestEncryption.java
@@ -63,6 +63,7 @@
import java.util.function.BooleanSupplier;
import javax.xml.bind.DatatypeConverter;
import org.apache.commons.lang3.RandomUtils;
+import org.apache.hadoop.crypto.CryptoInputStream;
import org.apache.hadoop.crypto.key.JavaKeyStoreProvider;
import org.apache.hadoop.crypto.key.KeyProvider;
import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
@@ -82,6 +83,7 @@
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import
org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB;
import org.apache.hadoop.hdds.scm.storage.MultipartInputStream;
+import org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper;
import
org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClientTestImpl;
import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.ClientConfigForTesting;
@@ -308,6 +310,17 @@ void testPutKeyWithEncryption(BucketLayout bucketLayout)
throws Exception {
createAndVerifyKeyData(bucket);
createAndVerifyStreamKeyData(bucket);
createAndVerifyFileSystemData(bucket);
+
+ String keyName = UUID.randomUUID().toString();
+ byte[] data = generateRandomData(PositionedReadTestHelper.SOURCE_SIZE);
+ DataTestUtil.createKey(bucket, keyName,
ReplicationConfig.fromTypeAndFactor(RATIS, ONE), data);
+ try (OzoneInputStream inputStream = bucket.readKey(keyName)) {
+ assertInstanceOf(CryptoInputStream.class, inputStream.getInputStream());
+ inputStream.seek(123);
+ PositionedReadTestHelper.runConcurrentPositionedReads(data,
inputStream::readFully);
+ assertEquals(123, inputStream.getPos());
+ assertEquals(Byte.toUnsignedInt(data[123]), inputStream.read());
+ }
}
@ParameterizedTest
@@ -721,6 +734,19 @@ private void testMultipartUploadWithEncryption(OzoneBucket
bucket,
try (OzoneInputStream inputStream = bucket.readKey(keyName)) {
assertInstanceOf(MultipartInputStream.class,
inputStream.getInputStream());
+ if (numParts == 2) {
+ inputStream.seek(123);
+ PositionedReadTestHelper.runConcurrentPositionedReads(inputData,
inputStream::readFully);
+ int offset = partsData.get(0).length - 17;
+ ByteBuffer destination =
ByteBuffer.allocateDirect(DEFAULT_CRYPTO_BUFFER_SIZE + 37);
+ inputStream.readFully(offset, destination);
+ destination.flip();
+ byte[] actual = new byte[destination.remaining()];
+ destination.get(actual);
+ assertArrayEquals(Arrays.copyOfRange(inputData, offset, offset +
actual.length), actual);
+ assertEquals(123, inputStream.getPos());
+ inputStream.seek(0);
+ }
// Test complete read
byte[] completeRead = new byte[keySize];
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java
index 3ce0af44527..1322b52ced7 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java
@@ -21,7 +21,7 @@
import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_BYTES_PER_CHECKSUM_MIN_SIZE;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.BufferedInputStream;
@@ -42,6 +42,7 @@
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
import org.apache.hadoop.hdds.scm.OzoneClientConfig;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.storage.BlockInputStream;
import org.apache.hadoop.hdds.scm.storage.StreamBlockInputStream;
import org.apache.hadoop.hdds.utils.IOUtils;
import org.apache.hadoop.hdds.utils.db.CodecBuffer;
@@ -193,7 +194,7 @@ void testUnevenFlushedChunks() throws Exception {
long shiftedOffset = chunks.get(1).getOffset();
assertTrue(shiftedOffset % config.getBytesPerChecksum() != 0, "second
chunk must shift the checksum grid");
try (KeyInputStream in = bucket.getKeyInputStream(key)) {
- assertTrue(in.isStreamBlockInputStream());
+ in.getPartStreams().forEach(part ->
assertInstanceOf(StreamBlockInputStream.class, part));
byte[] actual = new byte[expected.length];
org.apache.hadoop.io.IOUtils.readFully(in, actual, 0, actual.length);
assertArrayEquals(expected, actual);
@@ -296,7 +297,7 @@ private static void
assertClientBytesPerChecksum(SizeInBytes configured, OzoneCl
static void streamRead(SizeInBytes keySize, SizeInBytes bufferSize, String
expectedMD5,
BucketForTesting bucket, String keyName) throws Exception {
try (KeyInputStream in = bucket.getKeyInputStream(keyName)) {
- assertTrue(in.isStreamBlockInputStream());
+ in.getPartStreams().forEach(part ->
assertInstanceOf(StreamBlockInputStream.class, part));
runTestReadKey(keySize, bufferSize, expectedMD5, in);
}
}
@@ -305,7 +306,7 @@ static void nonStreamRead(SizeInBytes keySize, SizeInBytes
bufferSize, String ex
OzoneClient nonStreamReadClient, String volume, String bucket, String
keyName) throws Exception {
final ClientProtocol proxy = nonStreamReadClient.getProxy();
try (KeyInputStream in = (KeyInputStream) proxy.getKey(volume, bucket,
keyName).getInputStream()) {
- assertFalse(in.isStreamBlockInputStream());
+ in.getPartStreams().forEach(part ->
assertInstanceOf(BlockInputStream.class, part));
runTestReadKey(keySize, bufferSize, expectedMD5, in);
}
}
diff --git
a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/CapableOzoneFSInputStream.java
b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/CapableOzoneFSInputStream.java
index f2e94bf2508..c53f8b73aa5 100644
---
a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/CapableOzoneFSInputStream.java
+++
b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/CapableOzoneFSInputStream.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.fs.ozone;
import java.io.InputStream;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
import org.apache.hadoop.fs.FileSystem.Statistics;
import org.apache.hadoop.fs.StreamCapabilities;
import org.apache.hadoop.util.StringUtils;
@@ -34,8 +35,9 @@ public boolean hasCapability(String capability) {
switch (StringUtils.toLowerCase(capability)) {
case StreamCapabilities.READBYTEBUFFER:
case StreamCapabilities.UNBUFFER:
- case StreamCapabilities.PREADBYTEBUFFER:
return true;
+ case StreamCapabilities.PREADBYTEBUFFER:
+ return getWrappedInputStream() instanceof ByteBufferPositionedReadable;
default:
return false;
}
diff --git
a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java
b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java
index 1b8e5e9c9db..cb6e551d65d 100644
---
a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java
+++
b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java
@@ -30,7 +30,6 @@
import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.hdds.annotation.InterfaceAudience;
import org.apache.hadoop.hdds.annotation.InterfaceStability;
-import org.apache.hadoop.hdds.scm.storage.ExtendedInputStream;
import org.apache.hadoop.hdds.tracing.TracingUtil;
/**
@@ -39,10 +38,7 @@
* Sequential reads are NOT thread safe.
* <p>
* Positioned reads are thread safe.
- * When the underlying stream is an {@link ExtendedInputStream} and
- * when it supports {@link ExtendedInputStream#readFully(long, ByteBuffer)},
- * they delegate to it.
- * Otherwise, they fall back to the default synchronized seek-read-restore
implementation.
+ * They delegate to the underlying {@link ByteBufferPositionedReadable} stream.
* <p>
* Applications must use either sequential reads or position reads at any
given time,
* but not concurrent sequential/position reads
@@ -54,7 +50,6 @@ public class OzoneFSInputStream extends FSInputStream
private final InputStream inputStream;
private final Statistics statistics;
- private final Object positionedReadLock = new Object();
public OzoneFSInputStream(InputStream inputStream, Statistics statistics) {
this.inputStream = inputStream;
@@ -140,15 +135,15 @@ private int readInTrace(ByteBuffer buf) throws
IOException {
int readLen = Math.min(buf.remaining(), available());
if (buf.hasArray()) {
int pos = buf.position();
- bytesRead = read(buf.array(), pos, readLen);
+ bytesRead = inputStream.read(buf.array(), buf.arrayOffset() + pos,
readLen);
if (bytesRead > 0) {
buf.position(pos + bytesRead);
}
} else {
byte[] readData = new byte[readLen];
- bytesRead = read(readData, 0, readLen);
+ bytesRead = inputStream.read(readData, 0, readLen);
if (bytesRead > 0) {
- buf.put(readData);
+ buf.put(readData, 0, bytesRead);
}
}
}
@@ -175,31 +170,35 @@ public void unbuffer() {
*/
@Override
public int read(long position, ByteBuffer buffer) throws IOException {
- if (!buffer.hasRemaining()) {
- return 0;
- }
if (position < 0) {
throw new EOFException("position is negative: " + position);
}
- return readImpl(position, buffer);
+ if (buffer.isReadOnly()) {
+ throw new ReadOnlyBufferException();
+ }
+ return buffer.hasRemaining() ? readImpl(position, buffer) : 0;
}
@Override
public int read(long position, byte[] array, int offset, int length) throws
IOException {
- if (length == 0) {
- return 0;
- }
validatePositionedReadArgs(position, array, offset, length);
- return readImpl(position, ByteBuffer.wrap(array, offset, length));
+ return read(position, ByteBuffer.wrap(array, offset, length));
+ }
+
+ protected InputStream getWrappedInputStream() {
+ return inputStream;
}
private int readImpl(long position, ByteBuffer buffer) throws IOException {
- final Integer n = bestEffortExtendedInputStreamRead(position, buffer,
false);
- if (n != null) {
- return n;
+ if (!(inputStream instanceof ByteBufferPositionedReadable)) {
+ throw new UnsupportedOperationException("Positioned reads are not
supported by "
+ + inputStream.getClass().getName());
+ }
+ final int n = ((ByteBufferPositionedReadable) inputStream).read(position,
buffer);
+ if (statistics != null && n > 0) {
+ statistics.incrementBytesRead(n);
}
- // Fallback: stateful seek-read-restore on the shared cursor.
- return bestEffortReadFullySynchronized(position, buffer);
+ return n;
}
/**
@@ -210,20 +209,11 @@ private int readImpl(long position, ByteBuffer buffer)
throws IOException {
*/
@Override
public void readFully(long position, ByteBuffer buffer) throws IOException {
- if (!buffer.hasRemaining()) {
- return;
- }
- if (position < 0) {
- throw new EOFException("position is negative: " + position);
- }
readFullyImpl(position, buffer);
}
@Override
public void readFully(long position, byte[] array, int offset, int length)
throws IOException {
- if (length == 0) {
- return;
- }
validatePositionedReadArgs(position, array, offset, length);
readFullyImpl(position, ByteBuffer.wrap(array, offset, length));
}
@@ -234,94 +224,15 @@ public void readFully(long position, byte[] buffer)
throws IOException {
}
private void readFullyImpl(long position, ByteBuffer buffer) throws
IOException {
- final int length = buffer.remaining();
- final Integer n = bestEffortExtendedInputStreamRead(position, buffer,
true);
- if (n != null) {
- if (n < length) {
- throw new EOFException("Failed to read fully length " + length + " at
position " + position);
- }
- return;
- }
- final int m = bestEffortReadFullySynchronized(position, buffer);
- if (m < length) {
- throw new EOFException("Failed to read fully length " + length + " at
position " + position);
- }
- }
-
- /**
- * Best effort read via {@link ExtendedInputStream#readFully(long,
ByteBuffer)}.
- *
- * @return the number of bytes read; or null when the native stateless path
is unsupported.
- */
- private Integer bestEffortExtendedInputStreamRead(long position, ByteBuffer
buffer,
- boolean isReadFully) throws IOException {
- if (!(inputStream instanceof ExtendedInputStream)) {
- return null; // not an ExtendedInputStream
- }
- final int remainingBeforeRead = buffer.remaining();
- try {
- if (!((ExtendedInputStream) inputStream).readFully(position, buffer)) {
- return null; // ExtendedInputStream does not support readFully
+ do {
+ final int n = read(position, buffer);
+ if (n < 0) {
+ throw new EOFException("End of stream at position " + position);
}
- } catch (EOFException e) {
- if (isReadFully) {
- throw e;
+ if (n == 0 && buffer.hasRemaining()) {
+ throw new IOException("No progress reading at position " + position);
}
- // read() semantics: EOF -> -1 (fall through to bytesRead == 0 check
below)
- }
- final int bytesRead = remainingBeforeRead - buffer.remaining();
- if (bytesRead == 0) {
- return -1;
- }
- if (statistics != null) {
- statistics.incrementBytesRead(bytesRead);
- }
- return bytesRead;
- }
-
- /**
- * Fallback positioned read via synchronized seek-read-restore.
- * Uses {@link #read(ByteBuffer)} -- which handles streams that do not
implement
- * {@link org.apache.hadoop.fs.ByteBufferReadable} (e.g. GDPR
- * {@code javax.crypto.CipherInputStream}) -- so this works for any {@link
Seekable}
- * inner stream.
- *
- * @return the number of bytes read
- */
- private int bestEffortReadFullySynchronized(long position, ByteBuffer
buffer) throws IOException {
- synchronized (positionedReadLock) {
- final long oldPos = getPos();
- try {
- ((Seekable) inputStream).seek(position);
- final int n = bestEffortRead(buffer);
- if (statistics != null) {
- statistics.incrementBytesRead(n);
- }
- return n;
- } finally {
- ((Seekable) inputStream).seek(oldPos);
- }
- }
- }
-
- /**
- * Best effort read to fill up the buffer using {@link #read(ByteBuffer)}.
- *
- * @return the number of bytes read
- */
- private int bestEffortRead(ByteBuffer buffer) throws IOException {
- int readLength = 0;
- try {
- while (buffer.hasRemaining()) {
- final int n = read(buffer);
- if (n < 0) {
- return readLength;
- }
- readLength += n;
- }
- return readLength;
- } catch (EOFException e) {
- return readLength;
- }
+ position += n;
+ } while (buffer.hasRemaining());
}
}
diff --git
a/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
b/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
index 6954547e739..1caa968142a 100644
---
a/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
+++
b/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
@@ -20,6 +20,7 @@
import static
org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
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 static org.mockito.Mockito.anyString;
@@ -34,33 +35,34 @@
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
+import java.nio.ReadOnlyBufferException;
import java.security.GeneralSecurityException;
import java.util.Arrays;
import java.util.List;
import java.util.function.IntFunction;
+import javax.crypto.CipherInputStream;
+import javax.crypto.NullCipher;
import org.apache.commons.lang3.RandomUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.crypto.CipherSuite;
import org.apache.hadoop.crypto.CryptoCodec;
import org.apache.hadoop.crypto.CryptoInputStream;
import org.apache.hadoop.crypto.Decryptor;
+import org.apache.hadoop.fs.ByteBufferPositionedReadable;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.PositionedReadable;
import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.fs.StreamCapabilities;
-import org.apache.hadoop.hdds.scm.storage.ByteReaderStrategy;
-import org.apache.hadoop.hdds.scm.storage.ExtendedInputStream;
import org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper;
import org.apache.hadoop.ozone.client.io.KeyInputStream;
+import org.apache.hadoop.ozone.client.io.OzoneInputStream;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Timeout;
/**
* Tests for {@link OzoneFSInputStream}.
*/
public class TestOzoneFSInputStream {
- private static final byte CORRUPT_BYTE = (byte) 0x5A;
-
private static final List<IntFunction<ByteBuffer>> BUFFER_CONSTRUCTORS =
ImmutableList.of(ByteBuffer::allocate, ByteBuffer::allocateDirect);
@@ -198,193 +200,148 @@ public int read() {
}
@Test
- public void testByteBufferPositionedReadNegativePositionThrows() throws
Exception {
- // read(long, ByteBuffer) must throw EOFException for negative positions,
- // aligning with the byte-array PositionedReadable behaviour.
- final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
- final InterleavingSeekableInputStream underlying =
- new InterleavingSeekableInputStream(source);
- try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
- new FileSystem.Statistics("test"))) {
- ByteBuffer buf = ByteBuffer.allocate(16);
- assertThrows(EOFException.class, () -> subject.read(-1L, buf));
- }
- }
-
- @Test
- @Timeout(value = 30)
- public void testByteArrayFallbackWorksForSeekableOnlyStream() throws
Exception {
- // Regression: the old routing through read(long, ByteBuffer) would cast to
- // ByteBufferReadable in readAtPositionSeekRestore and throw
ClassCastException
- // for a stream that only implements Seekable (not ByteBufferReadable).
- final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
- final SeekableOnlyInputStream underlying = new
SeekableOnlyInputStream(source);
- try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
- new FileSystem.Statistics("test"))) {
- PositionedReadTestHelper.runConcurrentPositionedReads(source, (offset,
buf) -> {
- byte[] arr = new byte[buf.remaining()];
- subject.readFully(offset, arr);
- buf.put(arr);
- });
+ void cursorOnlyStreamsDoNotEmulatePositionedReads() throws Exception {
+ byte[] source = RandomUtils.secure().randomBytes(32);
+ try (CapableOzoneFSInputStream fs = new CapableOzoneFSInputStream(new
SeekableOnlyInputStream(source), null);
+ OzoneInputStream client = new OzoneInputStream(new
SeekableOnlyInputStream(source))) {
+ assertFalse(fs.hasCapability(StreamCapabilities.PREADBYTEBUFFER));
+ assertFalse(client.hasCapability(StreamCapabilities.PREADBYTEBUFFER));
+ assertFalse(client.hasCapability(StreamCapabilities.READBYTEBUFFER));
+ assertFalse(client.hasCapability(StreamCapabilities.UNBUFFER));
+ assertThrows(EOFException.class, () -> fs.read(-1,
ByteBuffer.allocate(1)));
+ assertThrows(UnsupportedOperationException.class, () -> fs.read(0,
ByteBuffer.allocate(1)));
+ assertThrows(UnsupportedOperationException.class, () -> client.read(0,
ByteBuffer.allocate(1)));
+ assertEquals(0, fs.getPos());
+ assertEquals(0, client.getPos());
+ assertEquals(Byte.toUnsignedInt(source[0]), fs.read());
+ assertEquals(Byte.toUnsignedInt(source[0]), client.read());
}
}
@Test
- @Timeout(value = 30)
- public void testConcurrentPositionedRead() throws Exception {
- final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
- final InterleavingSeekableInputStream underlying =
- new InterleavingSeekableInputStream(source);
- try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
- new FileSystem.Statistics("test"))) {
- PositionedReadTestHelper.runConcurrentPositionedReads(source,
- (offset, buf) -> subject.readFully(offset, buf));
+ void gdprDelegateDoesNotSupportPositionedReads() throws Exception {
+ byte[] data = new byte[] {1, 2, 3};
+ try (OzoneInputStream client = new OzoneInputStream(new
CipherInputStream(new ByteArrayInputStream(data),
+ new NullCipher()));
+ CapableOzoneFSInputStream fs = new
CapableOzoneFSInputStream(client.getInputStream(), null)) {
+ assertFalse(fs.hasCapability(StreamCapabilities.PREADBYTEBUFFER));
+ assertThrows(UnsupportedOperationException.class, () -> fs.read(0, new
byte[1], 0, 1));
+ byte[] actual = new byte[data.length];
+ assertEquals(data.length, fs.read(actual));
+ assertArrayEquals(data, actual);
}
}
@Test
- @Timeout(value = 30)
- public void testConcurrentPositionedReadEcFallback() throws Exception {
- final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
- final EcInterleavingInputStream underlying =
- new EcInterleavingInputStream(source);
- try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
- new FileSystem.Statistics("test"))) {
- PositionedReadTestHelper.runConcurrentPositionedReads(source,
- (offset, buf) -> subject.readFully(offset, buf));
+ void positionedReadsValidateEmptyRequests() throws Exception {
+ byte[] data = new byte[8];
+ try (OzoneInputStream client = new OzoneInputStream(new
NativePositionedInputStream(data));
+ OzoneFSInputStream fs = createTestSubject(client)) {
+ for (ByteBufferPositionedReadable stream : Arrays.asList(client, fs)) {
+ ByteBuffer empty = ByteBuffer.allocate(0);
+ ByteBuffer readOnly = empty.asReadOnlyBuffer();
+ assertThrows(EOFException.class, () -> stream.read(-1, empty));
+ assertThrows(EOFException.class, () -> stream.readFully(-1, empty));
+ assertThrows(ReadOnlyBufferException.class, () -> stream.read(0,
readOnly));
+ assertThrows(ReadOnlyBufferException.class, () -> stream.readFully(0,
readOnly));
+ assertThrows(NullPointerException.class, () -> stream.read(0, null));
+ assertThrows(NullPointerException.class, () -> stream.readFully(0,
null));
+ PositionedReadable arrayStream = (PositionedReadable) stream;
+ assertThrows(EOFException.class, () -> arrayStream.read(-1, data, 0,
0));
+ assertThrows(EOFException.class, () -> arrayStream.readFully(-1, data,
0, 0));
+ assertThrows(IllegalArgumentException.class, () -> arrayStream.read(0,
null, 0, 0));
+ assertThrows(IllegalArgumentException.class, () ->
arrayStream.readFully(0, null, 0, 0));
+ assertThrows(IndexOutOfBoundsException.class, () ->
arrayStream.read(0, data, -1, 0));
+ assertThrows(IndexOutOfBoundsException.class, () ->
arrayStream.readFully(0, data, -1, 0));
+ assertThrows(IndexOutOfBoundsException.class, () ->
arrayStream.read(0, data, data.length + 1, 0));
+ assertThrows(IndexOutOfBoundsException.class, () ->
arrayStream.readFully(0, data, data.length + 1, 0));
+ for (long position : new long[] {0, data.length, data.length + 1}) {
+ assertEquals(0, stream.read(position, empty));
+ stream.readFully(position, empty);
+ assertEquals(0, arrayStream.read(position, data, data.length, 0));
+ arrayStream.readFully(position, data, data.length, 0);
+ }
+ }
}
}
@Test
- @Timeout(value = 30)
- public void testConcurrentByteArrayPositionedReadEcFallback() throws
Exception {
- final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
- final EcInterleavingInputStream underlying =
- new EcInterleavingInputStream(source);
- try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
- new FileSystem.Statistics("test"))) {
- PositionedReadTestHelper.runConcurrentPositionedReads(source, (offset,
buf) -> {
- byte[] arr = new byte[buf.remaining()];
- subject.readFully(offset, arr);
- buf.put(arr);
- });
+ void clientRespectsDelegateCapabilities() throws Exception {
+ KeyInputStream key = mock(KeyInputStream.class);
+ when(key.hasCapability(anyString())).thenReturn(false);
+ try (OzoneInputStream client = new OzoneInputStream(key)) {
+ assertFalse(client.hasCapability(StreamCapabilities.PREADBYTEBUFFER));
+ assertFalse(client.hasCapability(StreamCapabilities.READBYTEBUFFER));
+ assertFalse(client.hasCapability(StreamCapabilities.UNBUFFER));
}
}
@Test
- @Timeout(value = 30)
- public void testConcurrentMixedApiEcFallback() throws Exception {
- // ByteBuffer and byte-array callers share positionedReadLock; verify no
interleaving.
- final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
- final EcInterleavingInputStream underlying =
- new EcInterleavingInputStream(source);
- try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
- new FileSystem.Statistics("test"))) {
- PositionedReadTestHelper.runConcurrentPositionedReads(source, (offset,
buf) -> {
+ void concurrentPositionedReadsThroughWrappers() throws Exception {
+ byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
+ try (OzoneFSInputStream stream = createTestSubject(new
OzoneInputStream(new NativePositionedInputStream(source)))) {
+ PositionedReadTestHelper.runConcurrentPositionedReads(source, (offset,
buffer) -> {
if ((offset & 1) == 0) {
- subject.readFully(offset, buf);
+ stream.readFully(offset, buffer);
} else {
- byte[] arr = new byte[buf.remaining()];
- subject.readFully(offset, arr);
- buf.put(arr);
+ byte[] bytes = new byte[buffer.remaining()];
+ stream.readFully(offset, bytes);
+ buffer.put(bytes);
}
});
}
}
- /**
- * Mimics KeyInputStream synchronized per-operation seek/read where
multi-steps
- * positioned reads must still be serialized at the FS layer.
- */
- private static final class InterleavingSeekableInputStream extends
InputStream
- implements Seekable, org.apache.hadoop.fs.ByteBufferReadable {
-
- private final InterleavingReadState readState;
-
- private InterleavingSeekableInputStream(byte[] data) {
- this.readState = new InterleavingReadState(data);
- }
-
- @Override
- public synchronized void seek(long p) {
- readState.seek(p);
- }
-
- @Override
- public synchronized long getPos() {
- return readState.getPos();
- }
-
- @Override
- public synchronized boolean seekToNewSource(long targetPos) {
- return false;
- }
-
- @Override
- public int read() {
- return -1;
- }
-
- @Override
- public synchronized int read(ByteBuffer buf) {
- return readState.read(buf);
+ @Test
+ void positionedReadDelegatesThroughWrappersAndCountsBytesOnce() throws
Exception {
+ byte[] source = RandomUtils.secure().randomBytes(32);
+ for (boolean wrapped : new boolean[] {false, true}) {
+ FileSystem.Statistics statistics = new FileSystem.Statistics("test");
+ InputStream input = new NativePositionedInputStream(source);
+ if (wrapped) {
+ input = new OzoneInputStream(input);
+ }
+ try (OzoneFSInputStream stream = new OzoneFSInputStream(input,
statistics)) {
+ byte[] result = new byte[12];
+ stream.readFully(3, result, 2, 7);
+ assertArrayEquals(Arrays.copyOfRange(source, 3, 10),
Arrays.copyOfRange(result, 2, 9));
+ assertEquals(7, statistics.getBytesRead());
+ ByteBuffer destination = ByteBuffer.wrap(result, 2, 7).slice();
+ assertEquals(3, stream.read(29, destination));
+ assertArrayEquals(Arrays.copyOfRange(source, 29, 32),
Arrays.copyOfRange(result, 2, 5));
+ assertEquals(10, statistics.getBytesRead());
+ assertEquals(-1, stream.read(32, ByteBuffer.allocateDirect(1)));
+ assertThrows(EOFException.class, () -> stream.readFully(31,
ByteBuffer.allocateDirect(2)));
+ assertEquals(11, statistics.getBytesRead());
+ }
}
}
- /**
- * Mimics an erasure-coded key stream: {@link ExtendedInputStream#readFully}
- * returns {@code false}, so {@link OzoneFSInputStream} falls back to
- * seek-read-restore on the shared cursor.
- */
- private static final class EcInterleavingInputStream extends
ExtendedInputStream {
-
- private final InterleavingReadState readState;
-
- private EcInterleavingInputStream(byte[] data) {
- this.readState = new InterleavingReadState(data);
- }
-
- @Override
- protected int readWithStrategy(ByteReaderStrategy strategy) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public synchronized void seek(long p) {
- readState.seek(p);
+ private static final class NativePositionedInputStream extends
ByteArrayInputStream
+ implements ByteBufferPositionedReadable {
+ private NativePositionedInputStream(byte[] data) {
+ super(data);
}
@Override
- public synchronized long getPos() {
- return readState.getPos();
- }
-
- @Override
- public synchronized boolean seekToNewSource(long targetPos) {
- return false;
- }
-
- @Override
- public synchronized int read(ByteBuffer buf) {
- return readState.read(buf);
- }
-
- @Override
- public synchronized int read(byte[] b, int off, int len) {
- return readState.read(b, off, len);
+ public int read(long position, ByteBuffer destination) {
+ if (position >= count) {
+ return -1;
+ }
+ int n = Math.min(destination.remaining(), count - (int) position);
+ destination.put(buf, (int) position, n);
+ return n;
}
@Override
- public void unbuffer() {
- return;
+ public void readFully(long position, ByteBuffer destination) throws
IOException {
+ int length = destination.remaining();
+ if (read(position, destination) < length) {
+ throw new EOFException();
+ }
}
}
- /**
- * A Seekable stream that does NOT implement ByteBufferReadable. Used to
verify
- * that the byte-array positioned-read fallback uses read(byte[]) rather than
- * casting to ByteBufferReadable (which would throw ClassCastException).
- */
private static final class SeekableOnlyInputStream extends InputStream
implements Seekable {
@@ -432,58 +389,4 @@ public boolean seekToNewSource(long targetPos) {
}
}
- private static final class InterleavingReadState {
- private final byte[] data;
- private long pos;
- private final ThreadLocal<Long> expectedReadPos = new ThreadLocal<>();
-
- private InterleavingReadState(byte[] data) {
- this.data = data;
- }
-
- private void seek(long p) {
- pos = p;
- expectedReadPos.set(p);
- }
-
- private long getPos() {
- return pos;
- }
-
- private int read(ByteBuffer buf) {
- Long expected = expectedReadPos.get();
- if (expected != null && pos != expected) {
- int len = buf.remaining();
- for (int i = 0; i < len; i++) {
- buf.put(CORRUPT_BYTE);
- }
- return len;
- }
- int toRead = Math.min(buf.remaining(), data.length - (int) pos);
- if (toRead <= 0) {
- return -1;
- }
- buf.put(data, (int) pos, toRead);
- pos += toRead;
- expectedReadPos.remove();
- return toRead;
- }
-
- private int read(byte[] b, int off, int len) {
- Long expected = expectedReadPos.get();
- if (expected != null && pos != expected) {
- java.util.Arrays.fill(b, off, off + len, CORRUPT_BYTE);
- return len;
- }
- int toRead = Math.min(len, data.length - (int) pos);
- if (toRead <= 0) {
- return -1;
- }
- System.arraycopy(data, (int) pos, b, off, toRead);
- pos += toRead;
- expectedReadPos.remove();
- return toRead;
- }
- }
-
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
