From c4b7e2365679a95f347629d46073bb695a3d54f4 Mon Sep 17 00:00:00 2001 From: "Tak Lon (Stephen) Wu" Date: Mon, 24 Aug 2026 10:31:40 -0700 Subject: [PATCH 1/4] HDDS-15424. Fix Concurrent positional read --- .../apache/hadoop/ozone/OzoneConfigKeys.java | 9 ++ .../hadoop/fs/ozone/BasicOzoneFileSystem.java | 10 +- .../fs/ozone/BasicRootedOzoneFileSystem.java | 10 +- .../fs/ozone/CapableOzoneFSInputStream.java | 7 +- .../hadoop/fs/ozone/OzoneFSInputStream.java | 25 +++- .../fs/ozone/TestOzoneFSInputStream.java | 135 ++++++++++++++++++ .../hadoop/fs/ozone/OzoneFileSystem.java | 3 +- .../fs/ozone/RootedOzoneFileSystem.java | 3 +- .../hadoop/fs/ozone/OzoneFileSystem.java | 3 +- .../fs/ozone/RootedOzoneFileSystem.java | 3 +- 10 files changed, 200 insertions(+), 8 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java index 87e6fb86f1ea..2d30d426d58b 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java @@ -131,6 +131,15 @@ public final class OzoneConfigKeys { public static final boolean OZONE_FS_HSYNC_ENABLED_DEFAULT = false; + /** + * When true, synchronize seek-read-restore in OzoneFSInputStream positioned + * reads for thread-safe pread on a shared input stream. + */ + public static final String OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED = + "ozone.fs.synchronize.positioned.reads.enabled"; + public static final boolean OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED_DEFAULT = + false; + /** * hsync lease soft limit. */ diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java index 7ade8339232c..c492013a6f16 100644 --- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java +++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java @@ -25,6 +25,8 @@ import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_LISTING_PAGE_SIZE; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_LISTING_PAGE_SIZE_DEFAULT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_MAX_LISTING_PAGE_SIZE; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED_DEFAULT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE_DEFAULT; import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_INDICATOR; @@ -258,7 +260,13 @@ public FSDataInputStream open(Path f, int bufferSize) throws IOException { } protected InputStream createFSInputStream(InputStream inputStream) { - return new OzoneFSInputStream(inputStream, statistics); + return new OzoneFSInputStream(inputStream, statistics, + isSynchronizePositionedReadsEnabled()); + } + + protected boolean isSynchronizePositionedReadsEnabled() { + return getConf().getBoolean(OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED, + OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED_DEFAULT); } protected void incrementCounter(Statistic statistic) { diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java index 089f0c453e6c..e075b9255c63 100644 --- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java +++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java @@ -25,6 +25,8 @@ import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_LISTING_PAGE_SIZE; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_LISTING_PAGE_SIZE_DEFAULT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_MAX_LISTING_PAGE_SIZE; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED_DEFAULT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE_DEFAULT; import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_INDICATOR; @@ -251,7 +253,13 @@ public FSDataInputStream open(Path path, int bufferSize) throws IOException { } protected InputStream createFSInputStream(InputStream inputStream) { - return new OzoneFSInputStream(inputStream, statistics); + return new OzoneFSInputStream(inputStream, statistics, + isSynchronizePositionedReadsEnabled()); + } + + protected boolean isSynchronizePositionedReadsEnabled() { + return getConf().getBoolean(OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED, + OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED_DEFAULT); } protected void incrementCounter(Statistic statistic) { 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 f2e94bf2508f..69afe35accb4 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 @@ -26,7 +26,12 @@ final class CapableOzoneFSInputStream extends OzoneFSInputStream implements StreamCapabilities { CapableOzoneFSInputStream(InputStream inputStream, Statistics statistics) { - super(inputStream, statistics); + this(inputStream, statistics, false); + } + + CapableOzoneFSInputStream(InputStream inputStream, Statistics statistics, + boolean synchronizePositionedReads) { + super(inputStream, statistics, synchronizePositionedReads); } @Override 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 a9c2c8b2f0fc..faf256a98a86 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 @@ -37,7 +37,9 @@ * The input stream for Ozone file system. * * TODO: Make inputStream generic for both rest and rpc clients - * This class is not thread safe. + * Sequential reads are not thread safe. Positioned reads may be made safe for + * concurrent pread when + * {@code ozone.fs.synchronize.positioned.reads.enabled=true}. */ @InterfaceAudience.Private @InterfaceStability.Evolving @@ -46,10 +48,18 @@ public class OzoneFSInputStream extends FSInputStream private final InputStream inputStream; private final Statistics statistics; + private final boolean synchronizePositionedReads; + private final Object positionedReadLock = new Object(); public OzoneFSInputStream(InputStream inputStream, Statistics statistics) { + this(inputStream, statistics, false); + } + + public OzoneFSInputStream(InputStream inputStream, Statistics statistics, + boolean synchronizePositionedReads) { this.inputStream = inputStream; this.statistics = statistics; + this.synchronizePositionedReads = synchronizePositionedReads; } @Override @@ -169,6 +179,15 @@ public int read(long position, ByteBuffer buf) throws IOException { if (!buf.hasRemaining()) { return 0; } + if (synchronizePositionedReads) { + synchronized (positionedReadLock) { + return readPositioned(position, buf); + } + } + return readPositioned(position, buf); + } + + private int readPositioned(long position, ByteBuffer buf) throws IOException { if (inputStream instanceof ExtendedInputStream) { final int remainingBeforeRead = buf.remaining(); try { @@ -179,7 +198,11 @@ public int read(long position, ByteBuffer buf) throws IOException { return -1; } } + return readAtPositionSeekRestore(position, buf); + } + private int readAtPositionSeekRestore(long position, ByteBuffer buf) + throws IOException { long oldPos = this.getPos(); int bytesRead; try { 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 86df63949b56..898b1110475d 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 @@ -19,6 +19,8 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.anyString; import static org.mockito.Mockito.mock; @@ -32,8 +34,14 @@ import java.io.InputStream; import java.nio.ByteBuffer; import java.security.GeneralSecurityException; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.function.IntFunction; import org.apache.commons.lang3.RandomUtils; import org.apache.hadoop.conf.Configuration; @@ -42,9 +50,13 @@ import org.apache.hadoop.crypto.CryptoInputStream; import org.apache.hadoop.crypto.Decryptor; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Seekable; import org.apache.hadoop.fs.StreamCapabilities; import org.apache.hadoop.ozone.client.io.KeyInputStream; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; /** * Tests for {@link OzoneFSInputStream}. @@ -187,4 +199,127 @@ public int read() { }; } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + @Timeout(value = 30) + public void testConcurrentPositionedRead(boolean synchronizePositionedReads) + throws Exception { + final byte[] source = RandomUtils.secure().randomBytes(512 * 1024); + final InterleavingSeekableInputStream underlying = + new InterleavingSeekableInputStream(source); + final OzoneFSInputStream subject = new OzoneFSInputStream(underlying, + new FileSystem.Statistics("test"), synchronizePositionedReads); + + if (synchronizePositionedReads) { + runConcurrentPositionedReads(subject, source); + } else { + ExecutionException executionException = assertThrows( + ExecutionException.class, + () -> runConcurrentPositionedReads(subject, source)); + assertInstanceOf(AssertionError.class, + unwrapExecutionException(executionException)); + } + } + + private static void runConcurrentPositionedReads(OzoneFSInputStream subject, + byte[] source) throws Exception { + ExecutorService pool = Executors.newFixedThreadPool(8); + try { + List> futures = new ArrayList<>(); + for (int t = 0; t < 8; t++) { + final int threadId = t; + futures.add(pool.submit(() -> { + try { + for (int i = 0; i < 100; i++) { + int offset = (threadId * 1000 + i * 17) % (source.length - 4096); + ByteBuffer buf = ByteBuffer.allocate(4096); + subject.readFully(offset, buf); + buf.flip(); + byte[] expected = Arrays.copyOfRange(source, offset, offset + 4096); + byte[] actual = new byte[4096]; + buf.get(actual); + assertArrayEquals(expected, actual, + "thread " + threadId + " offset " + offset); + } + } catch (IOException e) { + throw new RuntimeException(e); + } + })); + } + for (Future future : futures) { + future.get(1, TimeUnit.MINUTES); + } + } finally { + pool.shutdownNow(); + } + } + + private static Throwable unwrapExecutionException( + ExecutionException executionException) { + Throwable cause = executionException.getCause(); + while (cause instanceof RuntimeException && cause.getCause() != null) { + cause = cause.getCause(); + } + return cause; + } + + /** + * Mimics KeyInputStream synchronized per-operation seek/read where multi-step + * 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 static final byte CORRUPT_BYTE = (byte) 0x5A; + + private final byte[] data; + private long pos; + private final ThreadLocal expectedReadPos = new ThreadLocal<>(); + + private InterleavingSeekableInputStream(byte[] data) { + this.data = data; + } + + @Override + public synchronized void seek(long p) { + pos = p; + expectedReadPos.set(p); + } + + @Override + public synchronized long getPos() { + return pos; + } + + @Override + public synchronized boolean seekToNewSource(long targetPos) { + return false; + } + + @Override + public int read() { + return -1; + } + + @Override + public synchronized 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; + } + } + } diff --git a/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java b/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java index b23f0fb50877..db897d78d617 100644 --- a/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java +++ b/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java @@ -111,7 +111,8 @@ protected OzoneClientAdapter createAdapter(ConfigurationSource conf, @Override protected InputStream createFSInputStream(InputStream inputStream) { - return new CapableOzoneFSInputStream(inputStream, statistics); + return new CapableOzoneFSInputStream(inputStream, statistics, + isSynchronizePositionedReadsEnabled()); } @Override diff --git a/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java b/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java index 0031e57d31e5..a02e6083e124 100644 --- a/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java +++ b/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java @@ -109,7 +109,8 @@ protected OzoneClientAdapter createAdapter(ConfigurationSource conf, @Override protected InputStream createFSInputStream(InputStream inputStream) { - return new CapableOzoneFSInputStream(inputStream, statistics); + return new CapableOzoneFSInputStream(inputStream, statistics, + isSynchronizePositionedReadsEnabled()); } @Override diff --git a/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java b/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java index b23f0fb50877..db897d78d617 100644 --- a/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java +++ b/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java @@ -111,7 +111,8 @@ protected OzoneClientAdapter createAdapter(ConfigurationSource conf, @Override protected InputStream createFSInputStream(InputStream inputStream) { - return new CapableOzoneFSInputStream(inputStream, statistics); + return new CapableOzoneFSInputStream(inputStream, statistics, + isSynchronizePositionedReadsEnabled()); } @Override diff --git a/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java b/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java index 38dc72a77273..dc895f5af0b4 100644 --- a/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java +++ b/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java @@ -110,7 +110,8 @@ protected OzoneClientAdapter createAdapter(ConfigurationSource conf, @Override protected InputStream createFSInputStream(InputStream inputStream) { - return new CapableOzoneFSInputStream(inputStream, statistics); + return new CapableOzoneFSInputStream(inputStream, statistics, + isSynchronizePositionedReadsEnabled()); } @Override From d45d09dff3d14312e8398c442148f8e6c0612a81 Mon Sep 17 00:00:00 2001 From: "Tak Lon (Stephen) Wu" Date: Mon, 24 Aug 2026 11:52:14 -0700 Subject: [PATCH 2/4] Introduce stateless read for pread -align with HDDS-15920 --- .../hdds/scm/storage/BlockInputStream.java | 54 ++++- .../hdds/scm/storage/ChunkInputStream.java | 102 ++++++++++ .../scm/storage/LocalChunkInputStream.java | 7 + .../scm/storage/MultipartInputStream.java | 73 ++++++- .../scm/storage/DummyChunkInputStream.java | 18 +- .../scm/storage/PositionedReadTestHelper.java | 92 +++++++++ .../scm/storage/TestBlockInputStream.java | 43 ++++ .../scm/storage/TestChunkInputStream.java | 71 +++++++ .../scm/storage/TestMultipartInputStream.java | 126 ++++++++++++ .../src/main/resources/ozone-default.xml | 33 ++++ hadoop-ozone/ozonefs-common/pom.xml | 8 + .../hadoop/fs/ozone/OzoneFSInputStream.java | 23 ++- .../fs/ozone/TestOzoneFSInputStream.java | 185 ++++++++++++------ 13 files changed, 752 insertions(+), 83 deletions(-) create mode 100644 hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/PositionedReadTestHelper.java create mode 100644 hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestMultipartInputStream.java 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 9dcf7f66c26b..da2137a1a51c 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 @@ -23,6 +23,7 @@ import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -76,7 +77,7 @@ public class BlockInputStream extends BlockExtendedInputStream { private XceiverClientShortCircuit xceiverClientShortCircuit; private final AtomicBoolean fallbackToGrpc = new AtomicBoolean(false); private volatile FileInputStream blockFileInputStream; - private boolean initialized = false; + private volatile boolean initialized = false; // TODO: do we need to change retrypolicy based on exception. private final RetryPolicy retryPolicy; @@ -486,6 +487,57 @@ protected synchronized int readWithStrategy(ByteReaderStrategy strategy) * 2. chunkStream[2] will be seeked to position 10 * (= 90 - chunkOffset[2] (= 80)). */ + /** + * Stateless positioned read across this block's chunks. Fills up to + * {@code dst.remaining()} bytes starting from {@code blockRelativePosition} + * without mutating this stream's cursor ({@code chunkIndex}, + * {@code blockPosition}) or the chunk streams' buffered state, so it is safe + * for concurrent callers. Metadata ({@code chunkOffsets}, {@code + * chunkStreams}, {@code length}) is published once by {@link #initialize()}; + * {@link #initialized} is {@code volatile} so callers observe a consistent + * snapshot after initialization completes. + * + * @return bytes copied into {@code dst}, or {@link #EOF} at EOF + */ + int readPositioned(long blockRelativePosition, ByteBuffer dst) + throws IOException { + if (!initialized) { + initialize(); + } + final List streams = chunkStreams; + final long[] offsets = chunkOffsets; + final long blockLength = length; + if (streams == null || streams.isEmpty() + || blockRelativePosition < 0 || blockRelativePosition >= blockLength) { + return EOF; + } + + int total = 0; + long pos = blockRelativePosition; + while (dst.hasRemaining() && pos < blockLength) { + int idx = chunkIndexForPosition(pos, offsets); + ChunkInputStream chunk = streams.get(idx); + long chunkPos = pos - offsets[idx]; + int n = chunk.readPositioned(chunkPos, dst); + if (n <= 0) { + break; + } + total += n; + pos += n; + } + return total == 0 ? EOF : total; + } + + private static int chunkIndexForPosition(long pos, long[] offsets) { + int idx = Arrays.binarySearch(offsets, pos); + if (idx < 0) { + // binarySearch returns -insertionPoint - 1; the containing chunk is + // insertionPoint - 1. + idx = -idx - 2; + } + return idx; + } + @Override public synchronized void seek(long pos) throws IOException { if (!initialized) { 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 34ef7a71bd3f..85c56c6d0cdf 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 @@ -429,6 +429,108 @@ protected void readChunkDataIntoBuffers(ChunkInfo readChunkInfo) allocated = true; } + /** + * Whether this chunk stream can serve positioned reads without holding a + * lock. A plain chunk read is a self-contained RPC, so concurrent callers + * reading different ranges do not interfere. Overridden by + * {@link LocalChunkInputStream}, which reads from a shared {@link + * java.nio.channels.FileChannel} and therefore must serialize. + */ + boolean supportsConcurrentPositionedRead() { + return true; + } + + /** + * Stateless positioned read of up to {@code dst.remaining()} bytes starting + * at {@code chunkRelativePosition} within this chunk. Unlike the buffered + * {@link #read} path, this does not read or mutate any of the instance's + * buffer/position state ({@code buffers}, {@code chunkPosition}, + * {@code bufferOffsetWrtChunkData}, ...), so it is safe to call concurrently + * from multiple threads sharing the same stream. + * + * @param chunkRelativePosition start offset within this chunk + * @param dst destination buffer + * @return number of bytes copied into {@code dst}, or {@link #EOF} at EOF + */ + int readPositioned(long chunkRelativePosition, ByteBuffer dst) + throws IOException { + if (supportsConcurrentPositionedRead()) { + return doPositionedRead(chunkRelativePosition, dst); + } + // Local (short-circuit) reads share a FileChannel cursor; serialize them. + synchronized (this) { + return doPositionedRead(chunkRelativePosition, dst); + } + } + + private int doPositionedRead(long chunkRelativePosition, ByteBuffer dst) + throws IOException { + if (chunkRelativePosition < 0 || chunkRelativePosition >= length) { + return EOF; + } + final int toRead = + (int) Math.min(dst.remaining(), length - chunkRelativePosition); + if (toRead == 0) { + return 0; + } + + acquireClient(); + + final long adjustedOffset; + final long adjustedLen; + if (verifyChecksum) { + Pair boundaries = + computeChecksumBoundaries(chunkRelativePosition, toRead); + adjustedOffset = boundaries.getLeft(); + adjustedLen = boundaries.getRight(); + } else { + adjustedOffset = chunkRelativePosition; + adjustedLen = toRead; + } + + final ChunkInfo readChunkInfo = ChunkInfo.newBuilder(chunkInfo) + .setOffset(chunkInfo.getOffset() + adjustedOffset) + .setLen(adjustedLen) + .build(); + + final ByteBuffer[] readBuffers = readChunk(readChunkInfo); + return copyRange(readBuffers, chunkRelativePosition - adjustedOffset, + toRead, dst); + } + + /** + * Copy {@code toCopy} bytes from {@code src} buffers, skipping the first + * {@code skip} bytes, into {@code dst}. Operates on duplicates so the source + * buffers' positions are left untouched. + */ + private static int copyRange(ByteBuffer[] src, long skip, int toCopy, + ByteBuffer dst) { + long remainingSkip = skip; + int copied = 0; + for (ByteBuffer buffer : src) { + if (copied >= toCopy) { + break; + } + ByteBuffer dup = buffer.duplicate(); + if (remainingSkip > 0) { + int skipHere = (int) Math.min(remainingSkip, dup.remaining()); + dup.position(dup.position() + skipHere); + remainingSkip -= skipHere; + if (!dup.hasRemaining()) { + continue; + } + } + int n = Math.min(dup.remaining(), toCopy - copied); + if (n <= 0) { + continue; + } + dup.limit(dup.position() + n); + dst.put(dup); + copied += n; + } + return copied; + } + /** * Send RPC call to get the chunk from the container. */ diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java index 9de58ac7fe9a..d2ecdb07ea1b 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java @@ -72,6 +72,13 @@ public class LocalChunkInputStream extends ChunkInputStream } } + @Override + boolean supportsConcurrentPositionedRead() { + // Reads share a single FileChannel cursor, so positioned reads must be + // serialized rather than run concurrently. + return false; + } + /** * Get the chunk from the local block replica. */ 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 e7c259ce464a..95178564d44f 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 @@ -40,6 +40,7 @@ public class MultipartInputStream extends ExtendedInputStream { // List of PartInputStream, one for each part of the key private final List 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. @@ -70,15 +71,20 @@ public MultipartInputStream(String keyName, // 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; } @@ -191,10 +197,19 @@ public synchronized void seek(long pos) throws IOException { @Override public boolean readFully(long position, ByteBuffer buffer) throws IOException { - if (!isStreamBlockInputStream) { - return false; + if (isStreamBlockInputStream) { + return readFullyStreamBlock(position, buffer); } + return readFullyStateless(position, buffer); + } + /** + * Positioned read for the StreamBlock path. This emulates a positioned read + * with seek-read-restore on the shared stream cursor, so it is synchronized + * to keep concurrent callers from corrupting each other's position. + */ + private synchronized boolean readFullyStreamBlock(long position, + ByteBuffer buffer) throws IOException { final long oldPos = getPos(); seek(position); try { @@ -219,6 +234,60 @@ int readImpl(InputStream inputStream) throws IOException { return true; } + /** + * Stateless positioned read for the replicated (Ratis) path. Routes the read + * across the part {@link BlockInputStream}s using the immutable + * {@link #partOffsets} without seeking or mutating the shared cursor, so + * concurrent positioned reads run independently. Returns {@code false} (so + * the caller can fall back) when any part is not a {@link BlockInputStream}, + * e.g. erasure coded parts. + */ + private boolean readFullyStateless(long position, ByteBuffer buffer) + throws IOException { + if (!buffer.hasRemaining()) { + return true; + } + if (!statelessPositionedReadSupported) { + return false; + } + + 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); + } + int idx = partIndexForPosition(pos); + BlockInputStream part = (BlockInputStream) partStreams.get(idx); + long partPos = pos - partOffsets[idx]; + int n = part.readPositioned(partPos, buffer); + if (n <= 0) { + if (bytesRead > 0) { + return true; + } + throw new EOFException("EOF encountered at pos: " + pos + + " for key: " + key); + } + bytesRead += n; + pos += n; + } + return true; + } + + private int partIndexForPosition(long pos) { + int idx = Arrays.binarySearch(partOffsets, pos); + if (idx < 0) { + // binarySearch returns -insertionPoint - 1; the containing part is + // insertionPoint - 1. + idx = -idx - 2; + } + return idx; + } + public synchronized void initialize() throws IOException { // Pre-check that the stream has not been intialized already if (initialized) { diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java index 1ef6327c2a14..feeb72601fd0 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java @@ -19,7 +19,9 @@ import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Collections; import java.util.List; +import java.util.concurrent.atomic.AtomicReference; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; import org.apache.hadoop.hdds.scm.XceiverClientFactory; @@ -34,8 +36,9 @@ public class DummyChunkInputStream extends ChunkInputStream { private final byte[] chunkData; - // Stores the read chunk data in each readChunk call - private final List readByteBuffers = new ArrayList<>(); + // Buffers from the most recent readChunk call. + private final AtomicReference> lastReadByteBuffers = + new AtomicReference<>(); public DummyChunkInputStream(ChunkInfo chunkInfo, BlockID blockId, @@ -54,7 +57,7 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) { int bufferCapacity = readChunkInfo.getChecksumData().getBytesPerChecksum(); int bufferLen; - readByteBuffers.clear(); + List chunkBuffers = new ArrayList<>(); while (remainingToRead > 0) { if (remainingToRead < bufferCapacity) { bufferLen = remainingToRead; @@ -64,13 +67,15 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) { ByteString byteString = ByteString.copyFrom(chunkData, offset, bufferLen); - readByteBuffers.add(byteString); + chunkBuffers.add(byteString); offset += bufferLen; remainingToRead -= bufferLen; } - return BufferUtils.getReadOnlyByteBuffers(readByteBuffers) + lastReadByteBuffers.set(chunkBuffers); + + return BufferUtils.getReadOnlyByteBuffers(chunkBuffers) .toArray(new ByteBuffer[0]); } @@ -85,6 +90,7 @@ protected void releaseClient() { } public List getReadByteBuffers() { - return readByteBuffers; + List last = lastReadByteBuffers.get(); + return last != null ? last : Collections.emptyList(); } } 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 new file mode 100644 index 000000000000..fe5514698ffe --- /dev/null +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/PositionedReadTestHelper.java @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.hdds.scm.storage; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +/** + * Shared helpers for concurrent positioned-read unit tests. + */ +public final class PositionedReadTestHelper { + + public static final int THREAD_COUNT = 8; + public static final int ITERATIONS = 100; + public static final int BUFFER_SIZE = 4096; + public static final int SOURCE_SIZE = 512 * 1024; + + /** + * Performs a positioned read at the given key-relative offset into {@code buf}. + */ + @FunctionalInterface + public interface PositionedReadAction { + void readAtOffset(int offset, ByteBuffer buf) throws Exception; + } + + private PositionedReadTestHelper() { + } + + public static void runConcurrentPositionedReads(byte[] source, + PositionedReadAction action) throws Exception { + ExecutorService pool = Executors.newFixedThreadPool(THREAD_COUNT); + try { + List> futures = new ArrayList<>(); + for (int t = 0; t < THREAD_COUNT; t++) { + final int threadId = t; + futures.add(pool.submit((Callable) () -> { + for (int i = 0; i < ITERATIONS; i++) { + int offset = (threadId * 1000 + i * 17) % (source.length - BUFFER_SIZE); + ByteBuffer buf = ByteBuffer.allocate(BUFFER_SIZE); + action.readAtOffset(offset, buf); + buf.flip(); + byte[] expected = Arrays.copyOfRange(source, offset, offset + BUFFER_SIZE); + byte[] actual = new byte[BUFFER_SIZE]; + buf.get(actual); + assertArrayEquals(expected, actual, + "thread " + threadId + " offset " + offset); + } + return null; + })); + } + for (Future future : futures) { + future.get(1, TimeUnit.MINUTES); + } + } finally { + pool.shutdownNow(); + } + } + + public static Throwable unwrapExecutionException( + ExecutionException executionException) { + Throwable cause = executionException.getCause(); + while (cause instanceof RuntimeException && cause.getCause() != null) { + cause = cause.getCause(); + } + return cause; + } +} 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 1b5e7667e845..9b7b73285d17 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 @@ -18,6 +18,8 @@ package org.apache.hadoop.hdds.scm.storage; import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_NOT_FOUND; +import static org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.BUFFER_SIZE; +import static org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE; import static org.apache.hadoop.hdds.scm.storage.TestChunkInputStream.generateRandomData; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -469,4 +471,45 @@ private static Stream exceptionsTriggersRefresh() { new StatusException(Status.UNAVAILABLE)))) ); } + + @Test + public void testConcurrentPositionedRead() throws Exception { + try (BlockInputStream largeBlockStream = createLargeBlockStream()) { + largeBlockStream.initialize(); + PositionedReadTestHelper.runConcurrentPositionedReads(blockData, + (offset, buf) -> { + int read = largeBlockStream.readPositioned(offset, buf); + if (read != BUFFER_SIZE) { + throw new AssertionError("expected " + BUFFER_SIZE + " bytes at " + offset + + " but read " + read); + } + }); + } + } + + private BlockInputStream createLargeBlockStream() throws Exception { + refreshFunction = mock(Function.class); + BlockID blockID = new BlockID(new ContainerBlockID(1, 2)); + checksum = new Checksum(ChecksumType.NONE, 1024); + chunks = new ArrayList<>(1); + chunkDataMap = new HashMap<>(); + blockData = generateRandomData(SOURCE_SIZE); + blockSize = SOURCE_SIZE; + String chunkName = "large-chunk"; + ChunkInfo chunkInfo = ChunkInfo.newBuilder() + .setChunkName(chunkName) + .setOffset(0) + .setLen(SOURCE_SIZE) + .setChecksumData(checksum.computeChecksum( + blockData, 0, SOURCE_SIZE).getProtoBufMessage()) + .build(); + chunkDataMap.put(chunkName, blockData); + chunks.add(chunkInfo); + + OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); + clientConfig.setChecksumVerify(false); + Pipeline pipeline = MockPipeline.createSingleNodePipeline(); + return new DummyBlockInputStream(blockID, blockSize, pipeline, null, + null, refreshFunction, chunks, chunkDataMap, clientConfig); + } } 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 248ea8655223..96d911319f05 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 @@ -18,6 +18,8 @@ package org.apache.hadoop.hdds.scm.storage; import static org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.getReadChunkResponse; +import static org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.BUFFER_SIZE; +import static org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -156,6 +158,75 @@ public void testPartialChunkRead() throws Exception { matchWithInputData(chunkStream.getReadByteBuffers(), 0, 60); } + @Test + public void testPositionedReadIsStateless() throws Exception { + // Advance the stream's own cursor with a normal sequential read. + byte[] seq = new byte[10]; + assertEquals(10, chunkStream.read(seq, 0, 10)); + matchWithInputData(seq, 0, 10); + long posBeforePositionedRead = chunkStream.getPos(); + + // A positioned read at an unrelated offset returns the right bytes ... + ByteBuffer dst = ByteBuffer.allocate(15); + int read = chunkStream.readPositioned(50, dst); + assertEquals(15, read); + dst.flip(); + byte[] positioned = new byte[15]; + dst.get(positioned); + matchWithInputData(positioned, 50, 15); + + // ... without disturbing the stream's own position ... + assertEquals(posBeforePositionedRead, chunkStream.getPos()); + + // ... so the sequential read continues where it left off. + byte[] seq2 = new byte[10]; + assertEquals(10, chunkStream.read(seq2, 0, 10)); + matchWithInputData(seq2, 10, 10); + } + + @Test + 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)); + } + + @Test + public void testPositionedReadTruncatedAtChunkEnd() throws Exception { + // Request more than remains in the chunk; only the remainder is returned. + ByteBuffer dst = ByteBuffer.allocate(50); + int read = chunkStream.readPositioned(CHUNK_SIZE - 20, dst); + assertEquals(20, read); + dst.flip(); + byte[] tail = new byte[20]; + dst.get(tail); + matchWithInputData(tail, CHUNK_SIZE - 20, 20); + } + + @Test + public void testConcurrentPositionedRead() throws Exception { + Checksum checksum = new Checksum(ChecksumType.CRC32, BYTES_PER_CHECKSUM); + byte[] largeChunkData = generateRandomData(SOURCE_SIZE); + ChunkInfo largeChunkInfo = ChunkInfo.newBuilder() + .setChunkName("large-chunk") + .setOffset(0) + .setLen(SOURCE_SIZE) + .setChecksumData(checksum.computeChecksum( + largeChunkData, 0, SOURCE_SIZE).getProtoBufMessage()) + .build(); + try (DummyChunkInputStream largeStream = new DummyChunkInputStream( + largeChunkInfo, blockID, null, false, largeChunkData.clone(), null)) { + PositionedReadTestHelper.runConcurrentPositionedReads(largeChunkData, + (offset, buf) -> { + int read = largeStream.readPositioned(offset, buf); + if (read != BUFFER_SIZE) { + throw new AssertionError("expected " + BUFFER_SIZE + " bytes at " + offset + + " but read " + read); + } + }); + } + } + @Test public void testSeek() throws Exception { seekAndVerify(0); 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 new file mode 100644 index 000000000000..59a010c127f9 --- /dev/null +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestMultipartInputStream.java @@ -0,0 +1,126 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.hdds.scm.storage; + +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; + +import com.google.common.primitives.Bytes; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import org.apache.hadoop.hdds.client.BlockID; +import org.apache.hadoop.hdds.client.ContainerBlockID; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; +import org.apache.hadoop.hdds.scm.OzoneClientConfig; +import org.apache.hadoop.hdds.scm.pipeline.MockPipeline; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.ozone.common.Checksum; +import org.junit.jupiter.api.Test; + +/** + * Tests for {@link MultipartInputStream}'s functionality. + */ +public class TestMultipartInputStream { + + private static final int PART_SIZE = SOURCE_SIZE / 2; + + @Test + public void testPositionedReadPartialAtEof() throws Exception { + int fileLen = 1024; + byte[] partData = generateRandomData(fileLen); + OzoneConfiguration conf = new OzoneConfiguration(); + OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); + clientConfig.setChecksumVerify(false); + Pipeline pipeline = MockPipeline.createSingleNodePipeline(); + Function refreshFunction = mock(Function.class); + Checksum checksum = new Checksum(ChecksumType.NONE, 1024); + + BlockInputStream part = createBlockStream(new BlockID(new ContainerBlockID(1, 1)), + partData, pipeline, refreshFunction, clientConfig, checksum); + try (MultipartInputStream multipartStream = + new MultipartInputStream("test-key", Collections.singletonList(part))) { + multipartStream.initialize(); + int position = 12; + int expectedBytes = fileLen - position; + ByteBuffer buffer = ByteBuffer.allocate(expectedBytes * 2); + assertTrue(multipartStream.readFully(position, buffer)); + assertEquals(expectedBytes, buffer.position()); + } + } + + @Test + public void testConcurrentPositionedRead() throws Exception { + byte[] part0Data = generateRandomData(PART_SIZE); + byte[] part1Data = generateRandomData(PART_SIZE); + byte[] keyData = Bytes.concat(part0Data, part1Data); + + OzoneConfiguration conf = new OzoneConfiguration(); + OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); + clientConfig.setChecksumVerify(false); + Pipeline pipeline = MockPipeline.createSingleNodePipeline(); + Function refreshFunction = mock(Function.class); + Checksum checksum = new Checksum(ChecksumType.NONE, 1024); + + BlockInputStream part0 = createBlockStream(new BlockID(new ContainerBlockID(1, 1)), + part0Data, pipeline, refreshFunction, clientConfig, checksum); + BlockInputStream part1 = createBlockStream(new BlockID(new ContainerBlockID(1, 2)), + part1Data, pipeline, refreshFunction, clientConfig, checksum); + + List 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); + } + }); + } + } + + private static BlockInputStream createBlockStream(BlockID blockID, byte[] blockData, + Pipeline pipeline, Function refreshFunction, + OzoneClientConfig clientConfig, Checksum checksum) throws Exception { + List chunks = new ArrayList<>(1); + Map chunkDataMap = new HashMap<>(); + String chunkName = "chunk-" + blockID.getLocalID(); + ChunkInfo chunkInfo = ChunkInfo.newBuilder() + .setChunkName(chunkName) + .setOffset(0) + .setLen(blockData.length) + .setChecksumData(checksum.computeChecksum( + blockData, 0, blockData.length).getProtoBufMessage()) + .build(); + chunkDataMap.put(chunkName, blockData); + chunks.add(chunkInfo); + return new DummyBlockInputStream(blockID, blockData.length, pipeline, null, + null, refreshFunction, chunks, chunkDataMap, clientConfig); + } +} diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml index 9528ca27fc21..1850cae18ed6 100644 --- a/hadoop-hdds/common/src/main/resources/ozone-default.xml +++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml @@ -4715,6 +4715,39 @@ Can be enabled only when ozone.hbase.enhancements.allowed = true + + ozone.fs.synchronize.positioned.reads.enabled + false + OZONE, CLIENT + + Controls thread-safety of the fallback positioned-read + (ByteBufferPositionedReadable, e.g. HBase pread) path in + OzoneFSInputStream, where the read is emulated as a stateful + seek-read-restore on a single shared stream cursor. When multiple + threads issue positioned reads on the same shared input stream, that + sequence can interleave and return wrong bytes; setting this to true + serializes those positioned reads to make them correct. + + This flag only affects the fallback path. It does NOT apply to, and is + not needed for, the following: + 1. Replicated (Ratis) reads: positioned reads are served by a native, + stateless path that never touches the shared cursor. They are already + thread-safe and run in parallel with no lock, so this flag has no + effect on them. + 2. StreamBlock reads (ozone.client.stream.readblock.enable = true): + positioned reads are serialized internally regardless of this flag. + + When to enable: + - Set to true only when positioned reads fall back to the stateful path + and are issued concurrently on a shared stream - primarily erasure + coded (EC) buckets under concurrent pread. Enabling it + trades some pread parallelism (positioned reads on a stream are + serialized) for correctness. + Leave false (default) when: + - Buckets are replicated (Ratis); the native stateless path already + handles concurrency without serialization. + + ozone.om.lease.soft.limit 60s diff --git a/hadoop-ozone/ozonefs-common/pom.xml b/hadoop-ozone/ozonefs-common/pom.xml index d7d0bb3b1d32..8634d92f8c15 100644 --- a/hadoop-ozone/ozonefs-common/pom.xml +++ b/hadoop-ozone/ozonefs-common/pom.xml @@ -93,6 +93,14 @@ org.slf4j slf4j-api + + + + org.apache.ozone + hdds-client + test-jar + test + 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 faf256a98a86..0f2fc3287597 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 @@ -179,15 +179,11 @@ public int read(long position, ByteBuffer buf) throws IOException { if (!buf.hasRemaining()) { return 0; } - if (synchronizePositionedReads) { - synchronized (positionedReadLock) { - return readPositioned(position, buf); - } - } - return readPositioned(position, buf); - } - - private int readPositioned(long position, ByteBuffer buf) throws IOException { + // Prefer the native positioned read. For the replicated (Ratis) path this + // is stateless and thread-safe by construction, so it runs without any + // lock and lets concurrent positioned reads proceed in parallel. The + // StreamBlock path serializes internally. Streams that do not support a + // native positioned read (e.g. erasure coded) return false here. if (inputStream instanceof ExtendedInputStream) { final int remainingBeforeRead = buf.remaining(); try { @@ -198,6 +194,15 @@ private int readPositioned(long position, ByteBuffer buf) throws IOException { return -1; } } + + // Fallback: stateful seek-read-restore on the shared cursor. This is only + // thread-safe when enabled. When enabled, the lock covers the entire + // seek-read-restore sequence. + if (synchronizePositionedReads) { + synchronized (positionedReadLock) { + return readAtPositionSeekRestore(position, buf); + } + } return readAtPositionSeekRestore(position, buf); } 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 898b1110475d..9a8c3b0a5ee0 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 @@ -17,6 +17,8 @@ package org.apache.hadoop.fs.ozone; +import static org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE; +import static org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.unwrapExecutionException; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; @@ -34,14 +36,9 @@ import java.io.InputStream; import java.nio.ByteBuffer; import java.security.GeneralSecurityException; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; import java.util.function.IntFunction; import org.apache.commons.lang3.RandomUtils; import org.apache.hadoop.conf.Configuration; @@ -52,6 +49,9 @@ import org.apache.hadoop.fs.FileSystem; 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.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; @@ -63,6 +63,8 @@ */ public class TestOzoneFSInputStream { + private static final byte CORRUPT_BYTE = (byte) 0x5A; + private static final List> BUFFER_CONSTRUCTORS = ImmutableList.of(ByteBuffer::allocate, ByteBuffer::allocateDirect); @@ -204,91 +206,72 @@ public int read() { @Timeout(value = 30) public void testConcurrentPositionedRead(boolean synchronizePositionedReads) throws Exception { - final byte[] source = RandomUtils.secure().randomBytes(512 * 1024); + final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE); final InterleavingSeekableInputStream underlying = new InterleavingSeekableInputStream(source); - final OzoneFSInputStream subject = new OzoneFSInputStream(underlying, - new FileSystem.Statistics("test"), synchronizePositionedReads); - - if (synchronizePositionedReads) { - runConcurrentPositionedReads(subject, source); - } else { - ExecutionException executionException = assertThrows( - ExecutionException.class, - () -> runConcurrentPositionedReads(subject, source)); - assertInstanceOf(AssertionError.class, - unwrapExecutionException(executionException)); - } - } - - private static void runConcurrentPositionedReads(OzoneFSInputStream subject, - byte[] source) throws Exception { - ExecutorService pool = Executors.newFixedThreadPool(8); - try { - List> futures = new ArrayList<>(); - for (int t = 0; t < 8; t++) { - final int threadId = t; - futures.add(pool.submit(() -> { - try { - for (int i = 0; i < 100; i++) { - int offset = (threadId * 1000 + i * 17) % (source.length - 4096); - ByteBuffer buf = ByteBuffer.allocate(4096); - subject.readFully(offset, buf); - buf.flip(); - byte[] expected = Arrays.copyOfRange(source, offset, offset + 4096); - byte[] actual = new byte[4096]; - buf.get(actual); - assertArrayEquals(expected, actual, - "thread " + threadId + " offset " + offset); - } - } catch (IOException e) { - throw new RuntimeException(e); - } - })); - } - for (Future future : futures) { - future.get(1, TimeUnit.MINUTES); + try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying, + new FileSystem.Statistics("test"), synchronizePositionedReads)) { + + if (synchronizePositionedReads) { + PositionedReadTestHelper.runConcurrentPositionedReads(source, + (offset, buf) -> subject.readFully(offset, buf)); + } else { + ExecutionException executionException = assertThrows( + ExecutionException.class, + () -> PositionedReadTestHelper.runConcurrentPositionedReads(source, + (offset, buf) -> subject.readFully(offset, buf))); + assertInstanceOf(AssertionError.class, + unwrapExecutionException(executionException)); } - } finally { - pool.shutdownNow(); } } - private static Throwable unwrapExecutionException( - ExecutionException executionException) { - Throwable cause = executionException.getCause(); - while (cause instanceof RuntimeException && cause.getCause() != null) { - cause = cause.getCause(); + @ParameterizedTest + @ValueSource(booleans = {false, true}) + @Timeout(value = 30) + public void testConcurrentPositionedReadEcFallback(boolean synchronizePositionedReads) + 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"), synchronizePositionedReads)) { + + if (synchronizePositionedReads) { + PositionedReadTestHelper.runConcurrentPositionedReads(source, + (offset, buf) -> subject.readFully(offset, buf)); + } else { + ExecutionException executionException = assertThrows( + ExecutionException.class, + () -> PositionedReadTestHelper.runConcurrentPositionedReads(source, + (offset, buf) -> subject.readFully(offset, buf))); + assertInstanceOf(AssertionError.class, + unwrapExecutionException(executionException)); + } } - return cause; } /** - * Mimics KeyInputStream synchronized per-operation seek/read where multi-step + * 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 static final byte CORRUPT_BYTE = (byte) 0x5A; - - private final byte[] data; - private long pos; - private final ThreadLocal expectedReadPos = new ThreadLocal<>(); + private final InterleavingReadState readState; private InterleavingSeekableInputStream(byte[] data) { - this.data = data; + this.readState = new InterleavingReadState(data); } @Override public synchronized void seek(long p) { - pos = p; - expectedReadPos.set(p); + readState.seek(p); } @Override public synchronized long getPos() { - return pos; + return readState.getPos(); } @Override @@ -303,6 +286,78 @@ public int read() { @Override public synchronized int read(ByteBuffer buf) { + return readState.read(buf); + } + } + + /** + * 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 + public boolean readFully(long position, ByteBuffer buffer) { + return false; + } + + @Override + protected int readWithStrategy(ByteReaderStrategy strategy) { + throw new UnsupportedOperationException(); + } + + @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 synchronized int read(ByteBuffer buf) { + return readState.read(buf); + } + + @Override + public void unbuffer() { + return; + } + } + + private static final class InterleavingReadState { + private final byte[] data; + private long pos; + private final ThreadLocal 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(); From ff11fb61b390ce0ceddb24f510698bf212639ac5 Mon Sep 17 00:00:00 2001 From: "Tak Lon (Stephen) Wu" Date: Wed, 26 Aug 2026 14:32:45 -0700 Subject: [PATCH 3/4] address comments - use positional FileChannel reads for local read - fix retry with the same token when using block input stream - remove the opt-in flag, default to stateless pread for ExtendedInputStream and fallback to sync if it's all other input stream --- .../hdds/scm/storage/BlockInputStream.java | 37 ++++++++++- .../hdds/scm/storage/ChunkInputStream.java | 5 +- .../scm/storage/LocalChunkInputStream.java | 28 ++++++-- .../scm/storage/TestBlockInputStream.java | 66 +++++++++++++++++++ .../apache/hadoop/ozone/OzoneConfigKeys.java | 9 --- .../src/main/resources/ozone-default.xml | 33 ---------- .../hadoop/fs/ozone/BasicOzoneFileSystem.java | 10 +-- .../fs/ozone/BasicRootedOzoneFileSystem.java | 10 +-- .../fs/ozone/CapableOzoneFSInputStream.java | 7 +- .../hadoop/fs/ozone/OzoneFSInputStream.java | 26 +++----- .../fs/ozone/TestOzoneFSInputStream.java | 50 +++----------- .../hadoop/fs/ozone/OzoneFileSystem.java | 3 +- .../fs/ozone/RootedOzoneFileSystem.java | 3 +- .../hadoop/fs/ozone/OzoneFileSystem.java | 3 +- .../fs/ozone/RootedOzoneFileSystem.java | 3 +- 15 files changed, 153 insertions(+), 140 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 da2137a1a51c..9db331348a39 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 @@ -518,7 +518,7 @@ int readPositioned(long blockRelativePosition, ByteBuffer dst) int idx = chunkIndexForPosition(pos, offsets); ChunkInputStream chunk = streams.get(idx); long chunkPos = pos - offsets[idx]; - int n = chunk.readPositioned(chunkPos, dst); + int n = readChunkPositionedWithRetry(chunk, chunkPos, dst); if (n <= 0) { break; } @@ -528,6 +528,39 @@ int readPositioned(long blockRelativePosition, ByteBuffer dst) return total == 0 ? EOF : total; } + /** + * Positioned read for one chunk with the same retry and pipeline/token + * refresh handling as {@link #readWithStrategy(ByteReaderStrategy)}. + */ + private int readChunkPositionedWithRetry(ChunkInputStream chunk, long chunkPos, + ByteBuffer dst) throws IOException { + while (true) { + try { + int n = chunk.readPositioned(chunkPos, dst); + retries = 0; + return n; + } catch (SCMSecurityException ex) { + throw ex; + } catch (StorageContainerException e) { + if (shouldRetryRead(e, retryPolicy, ++retries)) { + handleReadError(e); + continue; + } + throw e; + } catch (IOException ex) { + if (shouldRetryRead(ex, retryPolicy, ++retries)) { + if (isConnectivityIssue(ex)) { + handleReadError(ex); + } else { + chunk.releaseClient(); + } + continue; + } + throw ex; + } + } + } + private static int chunkIndexForPosition(long pos, long[] offsets) { int idx = Arrays.binarySearch(offsets, pos); if (idx < 0) { @@ -690,7 +723,7 @@ private synchronized void storePosition() { blockPosition = getPos(); } - private void handleReadError(IOException cause) throws IOException { + private synchronized void handleReadError(IOException cause) throws IOException { releaseClient(); final List inputStreams = this.chunkStreams; if (inputStreams != null) { 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 85c56c6d0cdf..82121118ec88 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 @@ -433,8 +433,9 @@ protected void readChunkDataIntoBuffers(ChunkInfo readChunkInfo) * Whether this chunk stream can serve positioned reads without holding a * lock. A plain chunk read is a self-contained RPC, so concurrent callers * reading different ranges do not interfere. Overridden by - * {@link LocalChunkInputStream}, which reads from a shared {@link - * java.nio.channels.FileChannel} and therefore must serialize. + * {@link LocalChunkInputStream} uses positional {@link FileChannel} reads on + * the shared block channel, so concurrent callers on different chunks do not + * interfere. */ boolean supportsConcurrentPositionedRead() { return true; diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java index d2ecdb07ea1b..761e07ca33e2 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java @@ -74,9 +74,7 @@ public class LocalChunkInputStream extends ChunkInputStream @Override boolean supportsConcurrentPositionedRead() { - // Reads share a single FileChannel cursor, so positioned reads must be - // serialized rather than run concurrently. - return false; + return true; } /** @@ -89,12 +87,34 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) int bytesPerChecksum = chunkInfo.getChecksumData().getBytesPerChecksum(); final ByteBuffer[] buffers = BufferUtils.assignByteBuffers(readChunkInfo.getLen(), bytesPerChecksum); - dataIn.position(readChunkInfo.getOffset()).read(buffers); + readAtOffset(buffers, readChunkInfo.getOffset()); Arrays.stream(buffers).forEach(ByteBuffer::flip); validator.accept(Arrays.asList(buffers), readChunkInfo); return buffers; } + /** + * Read into {@code buffers} starting at {@code fileOffset} using positional + * {@link FileChannel} reads so concurrent callers on different chunks (which + * share the same underlying channel) do not stomp each other's cursor. + */ + private void readAtOffset(ByteBuffer[] buffers, long fileOffset) throws IOException { + long pos = fileOffset; + for (ByteBuffer buffer : buffers) { + while (buffer.hasRemaining()) { + int n = dataIn.read(buffer, pos); + if (n <= 0) { + if (buffer.hasRemaining()) { + throw new IOException("Failed to read chunk data at offset " + pos + + " for block chunk " + chunkInfo.getChunkName()); + } + break; + } + pos += n; + } + } + } + private void validateChunk(List bufferList, ChunkInfo readChunkInfo) throws OzoneChecksumException { if (verifyChecksum) { 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 9b7b73285d17..baf4d7fef090 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 @@ -28,6 +28,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.any; import static org.mockito.Mockito.anyInt; +import static org.mockito.Mockito.anyLong; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.reset; @@ -314,6 +315,71 @@ void refreshesPipelineOnReadFailure(IOException ex) throws Exception { id -> newBlockLocationInfo); } + @ParameterizedTest + @MethodSource("exceptionsTriggersRefresh") + void refreshesPipelineOnPositionedReadFailure(IOException ex) throws Exception { + Pipeline pipeline = MockPipeline.createSingleNodePipeline(); + BlockLocationInfo blockLocationInfo = mock(BlockLocationInfo.class); + when(blockLocationInfo.getPipeline()).thenReturn(pipeline); + Pipeline newPipeline = MockPipeline.createSingleNodePipeline(); + BlockLocationInfo newBlockLocationInfo = mock(BlockLocationInfo.class); + + testRefreshesPipelineOnPositionedReadFailure(ex, blockLocationInfo, + id -> newBlockLocationInfo); + + when(newBlockLocationInfo.getPipeline()).thenReturn(newPipeline); + testRefreshesPipelineOnPositionedReadFailure(ex, blockLocationInfo, + id -> blockLocationInfo); + + when(newBlockLocationInfo.getPipeline()).thenReturn(null); + testRefreshesPipelineOnPositionedReadFailure(ex, blockLocationInfo, + id -> newBlockLocationInfo); + } + + private void testRefreshesPipelineOnPositionedReadFailure(IOException ex, + BlockLocationInfo blockLocationInfo, + Function refreshPipelineFunction) + throws Exception { + BlockID blockID = new BlockID(new ContainerBlockID(1, 1)); + final int len = 200; + final ChunkInputStream stream = throwingChunkInputStreamForPositionedRead(ex, len, true); + + when(this.refreshFunction.apply(any())) + .thenAnswer(inv -> refreshPipelineFunction.apply(blockID)); + + try (BlockInputStream subject = createSubject(blockID, + blockLocationInfo.getPipeline(), stream)) { + subject.initialize(); + ByteBuffer buf = ByteBuffer.allocate(len); + int bytesRead = subject.readPositioned(0, buf); + assertEquals(len, bytesRead); + verify(this.refreshFunction).apply(blockID); + } finally { + reset(this.refreshFunction); + } + } + + private static ChunkInputStream throwingChunkInputStreamForPositionedRead( + IOException ex, int len, boolean succeedOnRetry) throws IOException { + final ChunkInputStream stream = mock(ChunkInputStream.class); + OngoingStubbing stubbing = + when(stream.readPositioned(anyLong(), any(ByteBuffer.class))) + .thenThrow(ex); + if (succeedOnRetry) { + stubbing.thenAnswer(invocation -> { + ByteBuffer buffer = invocation.getArgument(1); + int n = Math.min(len, buffer.remaining()); + for (int i = 0; i < n; i++) { + buffer.put((byte) 0); + } + return n; + }); + } + when(stream.getRemaining()).thenReturn((long) len); + when(stream.getLength()).thenReturn((long) len); + return stream; + } + private void testRefreshesPipelineOnReadFailure(IOException ex, BlockLocationInfo blockLocationInfo, Function refreshPipelineFunction) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java index 2d30d426d58b..87e6fb86f1ea 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java @@ -131,15 +131,6 @@ public final class OzoneConfigKeys { public static final boolean OZONE_FS_HSYNC_ENABLED_DEFAULT = false; - /** - * When true, synchronize seek-read-restore in OzoneFSInputStream positioned - * reads for thread-safe pread on a shared input stream. - */ - public static final String OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED = - "ozone.fs.synchronize.positioned.reads.enabled"; - public static final boolean OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED_DEFAULT = - false; - /** * hsync lease soft limit. */ diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml index 1850cae18ed6..9528ca27fc21 100644 --- a/hadoop-hdds/common/src/main/resources/ozone-default.xml +++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml @@ -4715,39 +4715,6 @@ Can be enabled only when ozone.hbase.enhancements.allowed = true - - ozone.fs.synchronize.positioned.reads.enabled - false - OZONE, CLIENT - - Controls thread-safety of the fallback positioned-read - (ByteBufferPositionedReadable, e.g. HBase pread) path in - OzoneFSInputStream, where the read is emulated as a stateful - seek-read-restore on a single shared stream cursor. When multiple - threads issue positioned reads on the same shared input stream, that - sequence can interleave and return wrong bytes; setting this to true - serializes those positioned reads to make them correct. - - This flag only affects the fallback path. It does NOT apply to, and is - not needed for, the following: - 1. Replicated (Ratis) reads: positioned reads are served by a native, - stateless path that never touches the shared cursor. They are already - thread-safe and run in parallel with no lock, so this flag has no - effect on them. - 2. StreamBlock reads (ozone.client.stream.readblock.enable = true): - positioned reads are serialized internally regardless of this flag. - - When to enable: - - Set to true only when positioned reads fall back to the stateful path - and are issued concurrently on a shared stream - primarily erasure - coded (EC) buckets under concurrent pread. Enabling it - trades some pread parallelism (positioned reads on a stream are - serialized) for correctness. - Leave false (default) when: - - Buckets are replicated (Ratis); the native stateless path already - handles concurrency without serialization. - - ozone.om.lease.soft.limit 60s diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java index c492013a6f16..7ade8339232c 100644 --- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java +++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java @@ -25,8 +25,6 @@ import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_LISTING_PAGE_SIZE; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_LISTING_PAGE_SIZE_DEFAULT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_MAX_LISTING_PAGE_SIZE; -import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED; -import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED_DEFAULT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE_DEFAULT; import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_INDICATOR; @@ -260,13 +258,7 @@ public FSDataInputStream open(Path f, int bufferSize) throws IOException { } protected InputStream createFSInputStream(InputStream inputStream) { - return new OzoneFSInputStream(inputStream, statistics, - isSynchronizePositionedReadsEnabled()); - } - - protected boolean isSynchronizePositionedReadsEnabled() { - return getConf().getBoolean(OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED, - OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED_DEFAULT); + return new OzoneFSInputStream(inputStream, statistics); } protected void incrementCounter(Statistic statistic) { diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java index e075b9255c63..089f0c453e6c 100644 --- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java +++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java @@ -25,8 +25,6 @@ import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_LISTING_PAGE_SIZE; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_LISTING_PAGE_SIZE_DEFAULT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_MAX_LISTING_PAGE_SIZE; -import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED; -import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED_DEFAULT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE_DEFAULT; import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_INDICATOR; @@ -253,13 +251,7 @@ public FSDataInputStream open(Path path, int bufferSize) throws IOException { } protected InputStream createFSInputStream(InputStream inputStream) { - return new OzoneFSInputStream(inputStream, statistics, - isSynchronizePositionedReadsEnabled()); - } - - protected boolean isSynchronizePositionedReadsEnabled() { - return getConf().getBoolean(OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED, - OZONE_FS_SYNCHRONIZE_POSITIONED_READS_ENABLED_DEFAULT); + return new OzoneFSInputStream(inputStream, statistics); } protected void incrementCounter(Statistic statistic) { 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 69afe35accb4..f2e94bf2508f 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 @@ -26,12 +26,7 @@ final class CapableOzoneFSInputStream extends OzoneFSInputStream implements StreamCapabilities { CapableOzoneFSInputStream(InputStream inputStream, Statistics statistics) { - this(inputStream, statistics, false); - } - - CapableOzoneFSInputStream(InputStream inputStream, Statistics statistics, - boolean synchronizePositionedReads) { - super(inputStream, statistics, synchronizePositionedReads); + super(inputStream, statistics); } @Override 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 0f2fc3287597..249e79aa7f60 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 @@ -37,9 +37,9 @@ * The input stream for Ozone file system. * * TODO: Make inputStream generic for both rest and rpc clients - * Sequential reads are not thread safe. Positioned reads may be made safe for - * concurrent pread when - * {@code ozone.fs.synchronize.positioned.reads.enabled=true}. + * Sequential reads are not thread safe. Positioned reads use a native + * stateless path when the underlying {@link ExtendedInputStream} supports it; + * otherwise they fall back to a synchronized seek-read-restore sequence. */ @InterfaceAudience.Private @InterfaceStability.Evolving @@ -48,18 +48,11 @@ public class OzoneFSInputStream extends FSInputStream private final InputStream inputStream; private final Statistics statistics; - private final boolean synchronizePositionedReads; private final Object positionedReadLock = new Object(); public OzoneFSInputStream(InputStream inputStream, Statistics statistics) { - this(inputStream, statistics, false); - } - - public OzoneFSInputStream(InputStream inputStream, Statistics statistics, - boolean synchronizePositionedReads) { this.inputStream = inputStream; this.statistics = statistics; - this.synchronizePositionedReads = synchronizePositionedReads; } @Override @@ -195,15 +188,12 @@ public int read(long position, ByteBuffer buf) throws IOException { } } - // Fallback: stateful seek-read-restore on the shared cursor. This is only - // thread-safe when enabled. When enabled, the lock covers the entire - // seek-read-restore sequence. - if (synchronizePositionedReads) { - synchronized (positionedReadLock) { - return readAtPositionSeekRestore(position, buf); - } + // Fallback: stateful seek-read-restore on the shared cursor. Synchronize the + // full sequence so concurrent positioned reads remain thread-safe when the + // native stateless path is unavailable (e.g. erasure coded keys). + synchronized (positionedReadLock) { + return readAtPositionSeekRestore(position, buf); } - return readAtPositionSeekRestore(position, buf); } private int readAtPositionSeekRestore(long position, ByteBuffer buf) 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 9a8c3b0a5ee0..8a79a5404ddb 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 @@ -18,11 +18,8 @@ package org.apache.hadoop.fs.ozone; import static org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE; -import static org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.unwrapExecutionException; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertInstanceOf; -import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.anyString; import static org.mockito.Mockito.mock; @@ -38,7 +35,6 @@ import java.security.GeneralSecurityException; import java.util.Arrays; import java.util.List; -import java.util.concurrent.ExecutionException; import java.util.function.IntFunction; import org.apache.commons.lang3.RandomUtils; import org.apache.hadoop.conf.Configuration; @@ -55,8 +51,6 @@ import org.apache.hadoop.ozone.client.io.KeyInputStream; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; /** * Tests for {@link OzoneFSInputStream}. @@ -201,53 +195,29 @@ public int read() { }; } - @ParameterizedTest - @ValueSource(booleans = {false, true}) + @Test @Timeout(value = 30) - public void testConcurrentPositionedRead(boolean synchronizePositionedReads) - throws Exception { + 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"), synchronizePositionedReads)) { - - if (synchronizePositionedReads) { - PositionedReadTestHelper.runConcurrentPositionedReads(source, - (offset, buf) -> subject.readFully(offset, buf)); - } else { - ExecutionException executionException = assertThrows( - ExecutionException.class, - () -> PositionedReadTestHelper.runConcurrentPositionedReads(source, - (offset, buf) -> subject.readFully(offset, buf))); - assertInstanceOf(AssertionError.class, - unwrapExecutionException(executionException)); - } + new FileSystem.Statistics("test"))) { + PositionedReadTestHelper.runConcurrentPositionedReads(source, + (offset, buf) -> subject.readFully(offset, buf)); } } - @ParameterizedTest - @ValueSource(booleans = {false, true}) + @Test @Timeout(value = 30) - public void testConcurrentPositionedReadEcFallback(boolean synchronizePositionedReads) - throws Exception { + 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"), synchronizePositionedReads)) { - - if (synchronizePositionedReads) { - PositionedReadTestHelper.runConcurrentPositionedReads(source, - (offset, buf) -> subject.readFully(offset, buf)); - } else { - ExecutionException executionException = assertThrows( - ExecutionException.class, - () -> PositionedReadTestHelper.runConcurrentPositionedReads(source, - (offset, buf) -> subject.readFully(offset, buf))); - assertInstanceOf(AssertionError.class, - unwrapExecutionException(executionException)); - } + new FileSystem.Statistics("test"))) { + PositionedReadTestHelper.runConcurrentPositionedReads(source, + (offset, buf) -> subject.readFully(offset, buf)); } } diff --git a/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java b/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java index db897d78d617..b23f0fb50877 100644 --- a/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java +++ b/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java @@ -111,8 +111,7 @@ protected OzoneClientAdapter createAdapter(ConfigurationSource conf, @Override protected InputStream createFSInputStream(InputStream inputStream) { - return new CapableOzoneFSInputStream(inputStream, statistics, - isSynchronizePositionedReadsEnabled()); + return new CapableOzoneFSInputStream(inputStream, statistics); } @Override diff --git a/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java b/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java index a02e6083e124..0031e57d31e5 100644 --- a/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java +++ b/hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java @@ -109,8 +109,7 @@ protected OzoneClientAdapter createAdapter(ConfigurationSource conf, @Override protected InputStream createFSInputStream(InputStream inputStream) { - return new CapableOzoneFSInputStream(inputStream, statistics, - isSynchronizePositionedReadsEnabled()); + return new CapableOzoneFSInputStream(inputStream, statistics); } @Override diff --git a/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java b/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java index db897d78d617..b23f0fb50877 100644 --- a/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java +++ b/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java @@ -111,8 +111,7 @@ protected OzoneClientAdapter createAdapter(ConfigurationSource conf, @Override protected InputStream createFSInputStream(InputStream inputStream) { - return new CapableOzoneFSInputStream(inputStream, statistics, - isSynchronizePositionedReadsEnabled()); + return new CapableOzoneFSInputStream(inputStream, statistics); } @Override diff --git a/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java b/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java index dc895f5af0b4..38dc72a77273 100644 --- a/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java +++ b/hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java @@ -110,8 +110,7 @@ protected OzoneClientAdapter createAdapter(ConfigurationSource conf, @Override protected InputStream createFSInputStream(InputStream inputStream) { - return new CapableOzoneFSInputStream(inputStream, statistics, - isSynchronizePositionedReadsEnabled()); + return new CapableOzoneFSInputStream(inputStream, statistics); } @Override From ccd16995cf43fa1768e909129113a54e5629da79 Mon Sep 17 00:00:00 2001 From: "Tak Lon (Stephen) Wu" Date: Thu, 27 Aug 2026 10:40:55 -0700 Subject: [PATCH 4/4] Fix positioned-read concurrency in BlockInputStream and OzoneCryptoInputStream. - Use ephemeral ChunkInputStreams with per-call retry counters for block pread so concurrent positioned reads do not share sequential retry state or chunk stream buffers. - Serialize cursor-moving and positioned reads on OzoneCryptoInputStream because CryptoInputStream is not thread-safe. --- .../hdds/scm/storage/BlockInputStream.java | 110 ++++++++++-------- .../client/io/OzoneCryptoInputStream.java | 90 +++++++++++++- 2 files changed, 152 insertions(+), 48 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 9db331348a39..c4301639c6e3 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 @@ -491,11 +491,9 @@ protected synchronized int readWithStrategy(ByteReaderStrategy strategy) * Stateless positioned read across this block's chunks. Fills up to * {@code dst.remaining()} bytes starting from {@code blockRelativePosition} * without mutating this stream's cursor ({@code chunkIndex}, - * {@code blockPosition}) or the chunk streams' buffered state, so it is safe - * for concurrent callers. Metadata ({@code chunkOffsets}, {@code - * chunkStreams}, {@code length}) is published once by {@link #initialize()}; - * {@link #initialized} is {@code volatile} so callers observe a consistent - * snapshot after initialization completes. + * {@code blockPosition}) or the sequential chunk streams' buffered state. + * Each covering chunk is read through an ephemeral {@link ChunkInputStream} + * closed as soon as its bytes have been copied. * * @return bytes copied into {@code dst}, or {@link #EOF} at EOF */ @@ -504,71 +502,89 @@ int readPositioned(long blockRelativePosition, ByteBuffer dst) if (!initialized) { initialize(); } - final List streams = chunkStreams; final long[] offsets = chunkOffsets; + final BlockData currentBlockData = blockData; final long blockLength = length; - if (streams == null || streams.isEmpty() + if (offsets == null || currentBlockData == null || blockRelativePosition < 0 || blockRelativePosition >= blockLength) { return EOF; } - int total = 0; + final List chunkInfos = currentBlockData.getChunksList(); + int index = Arrays.binarySearch(offsets, blockRelativePosition); + if (index < 0) { + index = -index - 2; + } + long pos = blockRelativePosition; - while (dst.hasRemaining() && pos < blockLength) { - int idx = chunkIndexForPosition(pos, offsets); - ChunkInputStream chunk = streams.get(idx); - long chunkPos = pos - offsets[idx]; - int n = readChunkPositionedWithRetry(chunk, chunkPos, dst); - if (n <= 0) { - break; + int totalReadLen = 0; + while (dst.hasRemaining() && pos < blockLength && index < chunkInfos.size()) { + final ChunkInfo chunkInfo = chunkInfos.get(index); + final long chunkOffset = pos - offsets[index]; + final long numBytesToRead = Math.min( + Math.min(dst.remaining(), chunkInfo.getLen() - chunkOffset), blockLength - pos); + if (numBytesToRead <= 0) { + index++; + continue; } - total += n; - pos += n; + final int numBytesRead = + readChunkAt(chunkInfo, chunkOffset, (int) numBytesToRead, dst); + totalReadLen += numBytesRead; + pos += numBytesRead; + index++; } - return total == 0 ? EOF : total; + return totalReadLen == 0 ? EOF : totalReadLen; } /** - * Positioned read for one chunk with the same retry and pipeline/token - * refresh handling as {@link #readWithStrategy(ByteReaderStrategy)}. + * Read {@code numBytesToRead} bytes starting at {@code chunkOffset} of the given chunk into {@code dst} + * through an ephemeral {@link ChunkInputStream}, retrying like {@link #readWithStrategy(ByteReaderStrategy)} + * but with a retry counter local to this call. */ - private int readChunkPositionedWithRetry(ChunkInputStream chunk, long chunkPos, - ByteBuffer dst) throws IOException { + private int readChunkAt(ChunkInfo chunkInfo, long chunkOffset, int numBytesToRead, ByteBuffer dst) + throws IOException { + final int startPosition = dst.position(); + int preadRetries = 0; while (true) { + final ChunkInputStream chunkStream = createChunkInputStream(chunkInfo); + final int numBytesRead; try { - int n = chunk.readPositioned(chunkPos, dst); - retries = 0; - return n; + final int oldLimit = dst.limit(); + try { + dst.limit(startPosition + numBytesToRead); + numBytesRead = chunkStream.readPositioned(chunkOffset, dst); + } finally { + dst.limit(oldLimit); + } } catch (SCMSecurityException ex) { throw ex; - } catch (StorageContainerException e) { - if (shouldRetryRead(e, retryPolicy, ++retries)) { - handleReadError(e); - continue; + } catch (StorageContainerException ex) { + if (!shouldRetryRead(ex, retryPolicy, ++preadRetries)) { + throw ex; } - throw e; + handleReadError(ex); + dst.position(startPosition); + continue; } catch (IOException ex) { - if (shouldRetryRead(ex, retryPolicy, ++retries)) { - if (isConnectivityIssue(ex)) { - handleReadError(ex); - } else { - chunk.releaseClient(); - } - continue; + if (!shouldRetryRead(ex, retryPolicy, ++preadRetries)) { + throw ex; } - throw ex; + if (isConnectivityIssue(ex)) { + handleReadError(ex); + } + dst.position(startPosition); + continue; + } finally { + chunkStream.close(); } - } - } - private static int chunkIndexForPosition(long pos, long[] offsets) { - int idx = Arrays.binarySearch(offsets, pos); - if (idx < 0) { - // binarySearch returns -insertionPoint - 1; the containing chunk is - // insertionPoint - 1. - idx = -idx - 2; + if (numBytesRead != numBytesToRead) { + throw new IOException(String.format( + "Inconsistent read for chunkName=%s length=%d numBytesToRead=%d numBytesRead=%d", + chunkInfo.getChunkName(), chunkInfo.getLen(), numBytesToRead, numBytesRead)); + } + return numBytesRead; } - return idx; } @Override 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 521c1d9816e6..e795587c70ec 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 @@ -18,10 +18,13 @@ package org.apache.hadoop.ozone.client.io; import com.google.common.base.Preconditions; +import java.io.EOFException; import java.io.IOException; +import java.nio.ByteBuffer; import org.apache.hadoop.crypto.CryptoCodec; import org.apache.hadoop.crypto.CryptoInputStream; import org.apache.hadoop.crypto.CryptoStreamUtils; +import org.apache.hadoop.fs.FSExceptionMessages; import org.apache.hadoop.hdds.scm.storage.PartInputStream; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,6 +39,8 @@ public class OzoneCryptoInputStream extends CryptoInputStream private static final Logger LOG = LoggerFactory.getLogger(OzoneCryptoInputStream.class); + private static final int EOF = -1; + private final long length; private final int bufferSize; private final String keyName; @@ -73,8 +78,13 @@ 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. + */ @Override - public int read(byte[] b, int off, int len) throws IOException { + public synchronized int read(byte[] b, int off, int len) throws IOException { // CryptoInputStream reads hadoop.security.crypto.buffer.size number of // bytes (default 8KB) at a time. This needs to be taken into account // in calculating the numBytesToRead. @@ -136,6 +146,84 @@ keyName, partIndex, getLength(), numBytesToRead, return numBytesRead; } + @Override + public synchronized int read(ByteBuffer buf) throws IOException { + return super.read(buf); + } + + @Override + public synchronized void seek(long pos) throws IOException { + super.seek(pos); + } + + @Override + public synchronized long getPos() throws IOException { + return super.getPos(); + } + + /** + * Positioned read. Decryption can only happen at the Crypto buffer boundaries, so this stream cannot + * read at an arbitrary position without moving its cursor. The read is therefore serialized against the + * other reads on this stream: the cursor is moved to {@code position}, the data is read through + * {@link #read(byte[], int, int)} (which does the Crypto buffer boundary adjustment) and the cursor is + * restored before the lock is released. + * + * @param position the position to read from. + * @param dst the buffer to read into. + * @return the number of bytes copied into {@code dst}, or -1 if no byte could be read. + */ + @Override + public synchronized int read(long position, ByteBuffer dst) throws IOException { + if (!dst.hasRemaining()) { + return 0; + } + if (position < 0 || position >= getLength()) { + return EOF; + } + + final long oldPos = getPos(); + final int numBytesToRead = (int) Math.min(dst.remaining(), getLength() - position); + final byte[] buffer = new byte[Math.min(numBytesToRead, getBufferSize())]; + Throwable failure = null; + try { + seek(position); + int totalReadLen = 0; + while (totalReadLen < numBytesToRead) { + final int numBytesRead = read(buffer, 0, + Math.min(buffer.length, numBytesToRead - totalReadLen)); + if (numBytesRead <= 0) { + break; + } + dst.put(buffer, 0, numBytesRead); + totalReadLen += numBytesRead; + } + return totalReadLen == 0 ? EOF : totalReadLen; + } catch (Throwable t) { + failure = t; + throw t; + } finally { + try { + seek(oldPos); + } catch (IOException e) { + if (failure == null) { + throw e; + } + failure.addSuppressed(e); + } + } + } + + @Override + public synchronized 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); + } + } + } + /** * Get number of bytes to read from the current stream based on the length * to be read, number of bytes remaining in the stream and the Crypto buffer