diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java index 75e483ad55ec..5064676c6415 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java @@ -102,6 +102,7 @@ public boolean hasCapability(String capability) { switch (StringUtils.toLowerCase(capability)) { case StreamCapabilities.READBYTEBUFFER: case StreamCapabilities.UNBUFFER: + case StreamCapabilities.VECTOREDIO: return true; default: return false; 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 a28658b1ebbf..c6f2b0426e26 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 @@ -26,7 +26,10 @@ import java.util.Collections; import java.util.List; import java.util.Objects; +import java.util.function.IntFunction; import org.apache.hadoop.fs.FSExceptionMessages; +import org.apache.hadoop.fs.FileRange; +import org.apache.hadoop.hdds.utils.VectoredReadUtils; import org.apache.ratis.util.Preconditions; /** @@ -261,6 +264,95 @@ public synchronized long skip(long n) throws IOException { return toSkip; } + /** + * Implements vectored read for multipart input stream. + * This method reads multiple byte ranges asynchronously, potentially + * from different underlying part streams. + * + * @param ranges list of file ranges to read + * @param allocate function to allocate ByteBuffer for each range + * @throws IOException if there is an error performing the reads + * @apiNote This method is synchronized to prevent race conditions from + * concurrent readVectored() calls on the same stream instance. + */ + public synchronized void readVectored( + List ranges, + IntFunction allocate + ) throws IOException { + checkOpen(); + if (!initialized) { + initialize(); + } + + // Save the initial position + final long initialPosition = getPos(); + + // Use common vectored read implementation + VectoredReadUtils.performVectoredRead( + ranges, + allocate, + (offset, buffer) -> readRangeData(offset, buffer, initialPosition) + ); + + // Restore position + seek(initialPosition); + } + + /** + * Helper method to read data for a specific range. + * Uses synchronized seeks to read data from the correct position. + * Reads data fully, handling partial reads in a loop. + * + * @param offset the starting offset in the stream + * @param buffer the buffer to read data into + * @param initialPosition the initial position to restore after reading + * @throws IOException if there is an error reading data + */ + private void readRangeData(long offset, ByteBuffer buffer, long initialPosition) + throws IOException { + synchronized (this) { + try { + seek(offset); + int totalBytesToRead = buffer.remaining(); + + // Read directly into buffer's backing array if available + byte[] readBuffer; + int arrayOffset; + if (buffer.hasArray()) { + readBuffer = buffer.array(); + arrayOffset = buffer.arrayOffset() + buffer.position(); + } else { + // Use temp array for direct ByteBuffers + readBuffer = new byte[totalBytesToRead]; + arrayOffset = 0; + } + + int totalBytesRead = 0; + // Read in a loop to handle partial reads + while (totalBytesRead < totalBytesToRead) { + int bytesRead = read(readBuffer, arrayOffset + totalBytesRead, + totalBytesToRead - totalBytesRead); + if (bytesRead < 0) { + throw new EOFException("End of file reached before reading fully. " + + "Requested: " + totalBytesToRead + ", Read: " + totalBytesRead); + } + totalBytesRead += bytesRead; + } + + // If we used a temp array, copy to buffer + if (!buffer.hasArray()) { + buffer.put(readBuffer, 0, totalBytesRead); + } else { + // Update buffer position + buffer.position(buffer.position() + totalBytesRead); + } + } finally { + // Restore position + seek(initialPosition); + } + } + } + @Override public synchronized void close() throws IOException { closed = true; diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.java new file mode 100644 index 000000000000..4b0d160b9fdf --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/VectoredReadUtils.java @@ -0,0 +1,168 @@ +/* + * 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.utils; + +import static java.util.Objects.requireNonNull; + +import java.io.EOFException; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.function.IntFunction; +import org.apache.hadoop.fs.FileRange; +import org.apache.hadoop.hdds.annotation.InterfaceAudience; + +/** + * Utility class for vectored read operations. + * Based on Hadoop's org.apache.hadoop.fs.VectoredReadUtils. + */ +@InterfaceAudience.Private +public final class VectoredReadUtils { + private VectoredReadUtils() { + // Utility class, no instances + } + + /** + * Functional interface for reading a range of data. + */ + @FunctionalInterface + public interface RangeReader { + /** + * Read data from the specified offset into the buffer. + * + * @param offset the offset to read from + * @param buffer the buffer to read into + * @throws IOException if an error occurs during reading + */ + void readRange(long offset, ByteBuffer buffer) throws IOException; + } + + /** + * Validate a single range. + * @param range range to validate. + * @return the range. + * @param range type + * @throws IllegalArgumentException the range length is negative + * @throws EOFException the range offset is negative + * @throws NullPointerException if the range is null. + */ + public static T validateRangeRequest(T range) + throws EOFException { + requireNonNull(range, "range is null"); + + if (range.getLength() < 0) { + throw new IllegalArgumentException("length is negative in " + range); + } + if (range.getOffset() < 0) { + throw new EOFException("position is negative in range " + range); + } + return range; + } + + /** + * Sort the input ranges by offset; no validation is done. + * @param input input ranges. + * @return a new list of the ranges, sorted by offset. + */ + public static List sortRangeList(List input) { + final List l = new ArrayList<>(input); + l.sort(Comparator.comparingLong(FileRange::getOffset)); + return l; + } + + /** + * Validate a list of ranges (including overlapping checks). + * Based on Hadoop's validateAndSortRanges. + * Two ranges overlap when the start offset of second is less than + * the end offset of first. End offset is calculated as start offset + length. + * + * @param ranges list of file ranges to validate + * @throws NullPointerException if ranges list is null or contains null elements + * @throws EOFException if any range has a negative offset + * @throws IllegalArgumentException if there are overlapping ranges or a range element is invalid + */ + public static void validateRanges(List ranges) throws IOException { + requireNonNull(ranges, "Null input list"); + + if (ranges.isEmpty()) { + return; + } + + if (ranges.size() == 1) { + validateRangeRequest(ranges.get(0)); + return; + } + + // Sort ranges to check for overlaps efficiently + List sortedRanges = sortRangeList(ranges); + FileRange prev = null; + for (final FileRange current : sortedRanges) { + validateRangeRequest(current); + if (prev != null) { + // Check for overlap: current start < prev end + if (current.getOffset() < prev.getOffset() + prev.getLength()) { + throw new IllegalArgumentException( + "Overlapping ranges " + prev + " and " + current); + } + } + prev = current; + } + } + + /** + * Common implementation for vectored read operations. + * Validates ranges and submits async read tasks for each range. + * + * @param ranges list of file ranges to read + * @param allocate function to allocate ByteBuffer for each range + * @param reader function that performs the actual read operation + * @throws IOException if there is an error during validation + */ + public static void performVectoredRead( + List ranges, + IntFunction allocate, + RangeReader reader) throws IOException { + // Validate ranges before processing + validateRanges(ranges); + // Perform vectored read using positioned read operations + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + if (result == null) { + result = new CompletableFuture<>(); + range.setData(result); + } + final CompletableFuture finalResult = result; + final long offset = range.getOffset(); + final int length = range.getLength(); + // Submit async read task for this range + CompletableFuture.runAsync(() -> { + try { + ByteBuffer buffer = allocate.apply(length); + reader.readRange(offset, buffer); + buffer.flip(); + finalResult.complete(buffer); + } catch (Exception e) { + finalResult.completeExceptionally(e); + } + }); + } + } +} diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java index 2366893ef562..1443079342b7 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java @@ -20,9 +20,13 @@ import java.io.IOException; import java.io.InputStream; import java.nio.ByteBuffer; +import java.util.List; +import java.util.function.IntFunction; import org.apache.hadoop.fs.ByteBufferReadable; import org.apache.hadoop.fs.CanUnbuffer; +import org.apache.hadoop.fs.FileRange; import org.apache.hadoop.fs.Seekable; +import org.apache.hadoop.hdds.scm.storage.MultipartInputStream; /** * OzoneInputStream is used to read data from Ozone. @@ -121,4 +125,26 @@ public boolean seekToNewSource(long targetPos) throws IOException { "underlying inputStream"); } } + + /** + * Read data from multiple byte ranges asynchronously. + * This allows reading multiple discontiguous ranges from the same file + * efficiently with a single API call. + * + * @param ranges list of file ranges to read + * @param allocate function to allocate ByteBuffer for each range + * @throws IOException if there is an error performing the reads + */ + public void readVectored( + List ranges, + IntFunction allocate + ) throws IOException { + if (inputStream instanceof MultipartInputStream) { + ((MultipartInputStream) inputStream).readVectored(ranges, allocate); + } else { + throw new UnsupportedOperationException("Vectored read is not supported " + + "on the underlying inputStream of type " + + inputStream.getClass().getName()); + } + } } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java new file mode 100644 index 000000000000..2bf3d59bcce9 --- /dev/null +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java @@ -0,0 +1,684 @@ +/* + * 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.fs.contract; + +import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_LENGTH; +import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY; +import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_VECTOR; +import static org.apache.hadoop.fs.StreamCapabilities.VECTOREDIO_BUFFERS_SLICED; +import static org.apache.hadoop.fs.contract.ContractTestUtils.VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS; +import static org.apache.hadoop.fs.contract.ContractTestUtils.assertDatasetEquals; +import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile; +import static org.apache.hadoop.fs.contract.ContractTestUtils.range; +import static org.apache.hadoop.fs.contract.ContractTestUtils.returnBuffersToPoolPostRead; +import static org.apache.hadoop.fs.contract.ContractTestUtils.validateVectoredReadResult; +import static org.apache.hadoop.io.Sizes.S_128K; +import static org.apache.hadoop.io.Sizes.S_4K; +import static org.apache.hadoop.test.LambdaTestUtils.intercept; +import static org.apache.hadoop.test.LambdaTestUtils.interceptFuture; +import static org.apache.hadoop.util.functional.FutureIO.awaitFuture; +import static org.junit.jupiter.api.Assertions.fail; + +import java.io.EOFException; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.IntFunction; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileRange; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.impl.TrackingByteBufferPool; +import org.apache.hadoop.io.ElasticByteBufferPool; +import org.apache.hadoop.io.WeakReferencedElasticByteBufferPool; +import org.apache.hadoop.util.concurrent.HadoopExecutors; +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedClass; +import org.junit.jupiter.params.provider.MethodSource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Test Vectored Reads. + *

+ * Both the original readVectored(allocator) and the readVectored(allocator, release) + * operations are tested. + */ +@ParameterizedClass(name = "buffer-{0}") +@MethodSource("params") +public abstract class AbstractContractVectoredReadTest extends AbstractFSContractTestBase { + + private static final Logger LOG = + LoggerFactory.getLogger(AbstractContractVectoredReadTest.class); + + private static final int DATASET_LEN = S_128K; + private static final byte[] DATASET = ContractTestUtils.dataset(DATASET_LEN, 'a', 32); + private static final String VECTORED_READ_FILE_NAME = "vectored_file.txt"; + + /** + * Buffer allocator for vector IO. + */ + private final IntFunction allocate; + + /** + * Buffer pool for vector IO. + */ + private final ElasticByteBufferPool pool = + new WeakReferencedElasticByteBufferPool(); + + private final String bufferType; + + private final boolean isDirect; + + /** + * Path to the vector file. + */ + private Path vectorPath; + + /** + * Counter of buffer releases. + * Because not all implementations release buffers on failures, + * this is not yet used in assertions. + */ + private final AtomicInteger bufferReleases = new AtomicInteger(); + + public static List params() { + return Arrays.asList("direct", "array"); + } + + protected AbstractContractVectoredReadTest(String bufferType) { + this.bufferType = bufferType; + this.isDirect = !"array".equals(bufferType); + this.allocate = size -> pool.getBuffer(isDirect, size); + } + + /** + * Get the buffer allocator. + * @return allocator function for vector IO. + */ + protected IntFunction getAllocate() { + return allocate; + } + + /** + * The buffer release operation. + */ + protected void release(ByteBuffer buffer) { + LOG.info("Released buffer {}", buffer); + bufferReleases.incrementAndGet(); + pool.putBuffer(buffer); + } + + /** + * Get the vector IO buffer pool. + * @return a pool. + */ + + protected ElasticByteBufferPool getPool() { + return pool; + } + + @BeforeEach + @Override + public void setup() throws Exception { + super.setup(); + vectorPath = path(VECTORED_READ_FILE_NAME); + FileSystem fs = getFileSystem(); + createFile(fs, vectorPath, true, DATASET); + } + + @AfterEach + @Override + public void teardown() throws Exception { + pool.release(); + super.teardown(); + } + + /** + * Open the vector file. + * @return the input stream. + * @throws IOException failure. + */ + protected FSDataInputStream openVectorFile() throws IOException { + return openVectorFile(getFileSystem()); + } + + /** + * Open the vector file. + * @param fs filesystem to use + * @return the input stream. + * @throws IOException failure. + */ + protected FSDataInputStream openVectorFile(final FileSystem fs) throws IOException { + return awaitFuture( + fs.openFile(vectorPath) + .opt(FS_OPTION_OPENFILE_LENGTH, DATASET_LEN) + .opt(FS_OPTION_OPENFILE_READ_POLICY, + FS_OPTION_OPENFILE_READ_POLICY_VECTOR) + .build()); + } + + @Test + public void testVectoredReadMultipleRanges() throws Exception { + List fileRanges = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + FileRange fileRange = FileRange.createFileRange(i * 100, 100); + fileRanges.add(fileRange); + } + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate, this::release); + CompletableFuture[] completableFutures = new CompletableFuture[fileRanges.size()]; + int i = 0; + for (FileRange res : fileRanges) { + completableFutures[i++] = res.getData(); + } + CompletableFuture combinedFuture = CompletableFuture.allOf(completableFutures); + combinedFuture.get(); + + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testVectoredReadAndReadFully() throws Exception { + List fileRanges = new ArrayList<>(); + range(fileRanges, 100, 100); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + byte[] readFullRes = new byte[100]; + in.readFully(100, readFullRes); + ByteBuffer vecRes = awaitFuture(fileRanges.get(0).getData()); + Assertions.assertThat(vecRes) + .describedAs("Result from vectored read and readFully must match") + .isEqualByComparingTo(ByteBuffer.wrap(readFullRes)); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testVectoredReadWholeFile() throws Exception { + describe("Read the whole file in one single vectored read"); + List fileRanges = new ArrayList<>(); + range(fileRanges, 0, DATASET_LEN); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + ByteBuffer vecRes = awaitFuture(fileRanges.get(0).getData()); + Assertions.assertThat(vecRes) + .describedAs("Result from vectored read and readFully must match") + .isEqualByComparingTo(ByteBuffer.wrap(DATASET)); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + /** + * As the minimum seek value is 4*1024,none of the below ranges + * will get merged. + */ + @Test + public void testDisjointRanges() throws Exception { + List fileRanges = new ArrayList<>(); + range(fileRanges, 0, 100); + range(fileRanges, 4_000 + 101, 100); + range(fileRanges, 16_000 + 101, 100); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate, this::release); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + /** + * As the minimum seek value is 4*1024, all the below ranges + * will get merged into one. + */ + @Test + public void testAllRangesMergedIntoOne() throws Exception { + List fileRanges = new ArrayList<>(); + final int length = 100; + range(fileRanges, 0, length); + range(fileRanges, 4_000 - length - 1, length); + range(fileRanges, 8_000 - length - 1, length); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + /** + * As the minimum seek value is 4*1024, the first three ranges will be + * merged into and other two will remain as it is. + */ + @Test + public void testSomeRangesMergedSomeUnmerged() throws Exception { + FileSystem fs = getFileSystem(); + List fileRanges = new ArrayList<>(); + range(fileRanges, 8 * 1024, 100); + range(fileRanges, 14 * 1024, 100); + range(fileRanges, 10 * 1024, 100); + range(fileRanges, 2 * 1024 - 101, 100); + range(fileRanges, 40 * 1024, 1024); + FileStatus fileStatus = fs.getFileStatus(path(VECTORED_READ_FILE_NAME)); + CompletableFuture builder = + fs.openFile(path(VECTORED_READ_FILE_NAME)) + .withFileStatus(fileStatus) + .build(); + try (FSDataInputStream in = builder.get()) { + in.readVectored(fileRanges, allocate, this::release); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + /** + * Most file systems won't support overlapping ranges. + * Currently, only Raw Local supports it. + */ + @Test + public void testOverlappingRanges() throws Exception { + if (!isSupported(VECTOR_IO_OVERLAPPING_RANGES)) { + verifyExceptionalVectoredRead( + getSampleOverlappingRanges(), + IllegalArgumentException.class); + } else { + try (FSDataInputStream in = openVectorFile()) { + List fileRanges = getSampleOverlappingRanges(); + in.readVectored(fileRanges, allocate); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + } + + /** + * Same ranges are special case of overlapping. + */ + @Test + public void testSameRanges() throws Exception { + if (!isSupported(VECTOR_IO_OVERLAPPING_RANGES)) { + verifyExceptionalVectoredRead( + getSampleSameRanges(), + IllegalArgumentException.class); + } else { + try (FSDataInputStream in = openVectorFile()) { + List fileRanges = getSampleSameRanges(); + in.readVectored(fileRanges, allocate, this::release); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + } + + /** + * A null range is not permitted. + */ + @Test + public void testNullRange() throws Exception { + List fileRanges = new ArrayList<>(); + range(fileRanges, 500, 100); + fileRanges.add(null); + verifyExceptionalVectoredRead( + fileRanges, + NullPointerException.class); + } + + /** + * A null range is not permitted. + */ + @Test + public void testNullRangeList() throws Exception { + verifyExceptionalVectoredRead( + null, + NullPointerException.class); + } + + @Test + public void testSomeRandomNonOverlappingRanges() throws Exception { + List fileRanges = new ArrayList<>(); + range(fileRanges, 500, 100); + range(fileRanges, 1000, 200); + range(fileRanges, 50, 10); + range(fileRanges, 10, 5); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testConsecutiveRanges() throws Exception { + List fileRanges = new ArrayList<>(); + final int offset = 500; + final int length = 2011; + range(fileRanges, offset, length); + range(fileRanges, offset + length, length); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate, this::release); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testEmptyRanges() throws Exception { + List fileRanges = new ArrayList<>(); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + Assertions.assertThat(fileRanges) + .describedAs("Empty ranges must stay empty") + .isEmpty(); + } + } + + /** + * Test to validate EOF ranges. + *

+ * Default implementation fails with EOFException + * while reading the ranges. Some implementation like s3, checksum fs fail fast + * as they already have the file length calculated. + * The contract option {@link ContractOptions#VECTOR_IO_EARLY_EOF_CHECK} is used + * to determine which check to perform. + */ + @Test + public void testEOFRanges() throws Exception { + describe("Testing reading with an offset past the end of the file"); + List fileRanges = range(DATASET_LEN + 1, 100); + + if (isSupported(VECTOR_IO_EARLY_EOF_CHECK)) { + LOG.info("Expecting early EOF failure"); + verifyExceptionalVectoredRead(fileRanges, EOFException.class); + } else { + expectEOFinRead(fileRanges); + } + } + + @Test + public void testVectoredReadWholeFilePlusOne() throws Exception { + describe("Try to read whole file plus 1 byte"); + List fileRanges = range(0, DATASET_LEN + 1); + + if (isSupported(VECTOR_IO_EARLY_EOF_CHECK)) { + LOG.info("Expecting early EOF failure"); + verifyExceptionalVectoredRead(fileRanges, EOFException.class); + } else { + expectEOFinRead(fileRanges); + } + } + + private void expectEOFinRead(final List fileRanges) throws Exception { + LOG.info("Expecting late EOF failure"); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate, this::release); + for (FileRange res : fileRanges) { + CompletableFuture data = res.getData(); + interceptFuture(EOFException.class, + "", + VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, + TimeUnit.SECONDS, + data); + } + } + } + + @Test + public void testNegativeLengthRange() throws Exception { + verifyExceptionalVectoredRead(range(0, -50), IllegalArgumentException.class); + } + + @Test + public void testNegativeOffsetRange() throws Exception { + verifyExceptionalVectoredRead(range(-1, 50), EOFException.class); + } + + @Test + public void testNullReleaseOperation() throws Exception { + + final List range = range(0, 10); + try (FSDataInputStream in = openVectorFile()) { + intercept(NullPointerException.class, () -> + in.readVectored(range, allocate, null)); + } + } + + @Test + public void testNormalReadAfterVectoredRead() throws Exception { + List fileRanges = createSampleNonOverlappingRanges(); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, allocate); + // read starting 200 bytes + final int len = 200; + byte[] res = new byte[len]; + in.readFully(res, 0, len); + ByteBuffer buffer = ByteBuffer.wrap(res); + assertDatasetEquals(0, "normal_read", buffer, len, DATASET); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testVectoredReadAfterNormalRead() throws Exception { + List fileRanges = createSampleNonOverlappingRanges(); + try (FSDataInputStream in = openVectorFile()) { + // read starting 200 bytes + final int len = 200; + byte[] res = new byte[len]; + in.readFully(res, 0, len); + ByteBuffer buffer = ByteBuffer.wrap(res); + assertDatasetEquals(0, "normal_read", buffer, len, DATASET); + in.readVectored(fileRanges, allocate); + validateVectoredReadResult(fileRanges, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges, pool); + } + } + + @Test + public void testMultipleVectoredReads() throws Exception { + List fileRanges1 = createSampleNonOverlappingRanges(); + List fileRanges2 = createSampleNonOverlappingRanges(); + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges1, allocate); + in.readVectored(fileRanges2, allocate, this::release); + validateVectoredReadResult(fileRanges2, DATASET, 0); + validateVectoredReadResult(fileRanges1, DATASET, 0); + returnBuffersToPoolPostRead(fileRanges1, pool); + returnBuffersToPoolPostRead(fileRanges2, pool); + } + } + + /** + * This test creates list of ranges and then submit a readVectored + * operation and then uses a separate thread pool to process the + * results asynchronously. + */ + @Test + public void testVectoredIOEndToEnd() throws Exception { + List fileRanges = new ArrayList<>(); + range(fileRanges, 8 * 1024, 100); + range(fileRanges, 14 * 1024, 100); + range(fileRanges, 10 * 1024, 100); + range(fileRanges, 2 * 1024 - 101, 100); + range(fileRanges, 40 * 1024, 1024); + + ExecutorService dataProcessor = Executors.newFixedThreadPool(5); + CountDownLatch countDown = new CountDownLatch(fileRanges.size()); + + try (FSDataInputStream in = openVectorFile()) { + in.readVectored(fileRanges, this.allocate); + for (FileRange res : fileRanges) { + dataProcessor.submit(() -> { + try { + readBufferValidateDataAndReturnToPool(res, countDown); + } catch (Exception e) { + String error = String.format("Error while processing result for %s", res); + LOG.error(error, e); + fail(error, e); + } + }); + } + // user can perform other computations while waiting for IO. + if (!countDown.await(VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + fail("Timeout/Error while processing vectored io results"); + } + } finally { + HadoopExecutors.shutdown(dataProcessor, LOG, + VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + } + + private void readBufferValidateDataAndReturnToPool(FileRange res, + CountDownLatch countDownLatch) + throws IOException, TimeoutException { + try { + CompletableFuture data = res.getData(); + // Read the data and perform custom operation. Here we are just + // validating it with original data. + awaitFuture(data.thenAccept(buffer -> { + assertDatasetEquals((int) res.getOffset(), + "vecRead", buffer, res.getLength(), DATASET); + // return buffer to the pool once read. + // If the read failed, this doesn't get invoked. + pool.putBuffer(buffer); + }), + VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } finally { + // countdown to notify main thread that processing has been done. + countDownLatch.countDown(); + } + } + + protected List createSampleNonOverlappingRanges() { + List fileRanges = new ArrayList<>(); + range(fileRanges, 0, 100); + range(fileRanges, 110, 50); + return fileRanges; + } + + protected List getSampleSameRanges() { + List fileRanges = new ArrayList<>(); + range(fileRanges, 8_000, 1000); + range(fileRanges, 8_000, 1000); + range(fileRanges, 8_000, 1000); + return fileRanges; + } + + protected List getSampleOverlappingRanges() { + List fileRanges = new ArrayList<>(); + range(fileRanges, 100, 500); + range(fileRanges, 400, 500); + return fileRanges; + } + + protected List getConsecutiveRanges() { + List fileRanges = new ArrayList<>(); + range(fileRanges, 100, 500); + range(fileRanges, 600, 500); + return fileRanges; + } + + /** + * Validate that exceptions must be thrown during a vectored + * read operation with specific input ranges. + * @param fileRanges input file ranges. + * @param clazz type of exception expected. + * @throws Exception any other exception. + */ + protected void verifyExceptionalVectoredRead( + List fileRanges, + Class clazz) throws Exception { + + try (FSDataInputStream in = openVectorFile()) { + intercept(clazz, () -> { + in.readVectored(fileRanges, allocate); + return "triggered read of " + fileRanges.size() + " ranges" + " against " + in; + }); + } + } + + @Test + public void testBufferSlicing() throws Throwable { + describe("Test buffer slicing behavior in vectored IO"); + + final int numBuffers = 8; + final int bufferSize = S_4K; + long offset = 0; + final List fileRanges = new ArrayList<>(); + for (int i = 0; i < numBuffers; i++) { + fileRanges.add(FileRange.createFileRange(offset, bufferSize)); + // increment and add a non-binary-aligned gap, so as to force + // offsets to be misaligned with possible page sizes. + offset += bufferSize + 4000; + } + TrackingByteBufferPool trackerPool = TrackingByteBufferPool.wrap(getPool()); + int unknownBuffers = 0; + boolean slicing; + try (FSDataInputStream in = openVectorFile()) { + slicing = in.hasCapability(VECTOREDIO_BUFFERS_SLICED); + LOG.info("Slicing is {} for vectored IO with stream {}", slicing, in); + in.readVectored(fileRanges, s -> trackerPool.getBuffer(isDirect, s), trackerPool::putBuffer); + + // check that all buffers are from the the pool, unless they are sliced. + for (FileRange res : fileRanges) { + CompletableFuture data = res.getData(); + ByteBuffer buffer = awaitFuture(data); + Assertions.assertThat(buffer) + .describedAs("Buffer must not be null") + .isNotNull(); + Assertions.assertThat(slicing || trackerPool.containsBuffer(buffer)) + .describedAs("Buffer must be from the pool") + .isTrue(); + try { + trackerPool.putBuffer(buffer); + } catch (TrackingByteBufferPool.ReleasingUnallocatedByteBufferException e) { + // this can happen if the buffer was sliced, as it is not in the pool. + if (!slicing) { + throw e; + } + LOG.info("Sliced buffer detected: {}", buffer); + unknownBuffers++; + } + } + } + try { + trackerPool.close(); + } catch (TrackingByteBufferPool.LeakedByteBufferException e) { + if (!slicing) { + throw e; + } + LOG.info("Slicing is enabled; we saw leaked buffers: {} after {}" + + " releases of unknown buffers", + e.getCount(), unknownBuffers); + } + + } +} diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java index 1ddaa079bb95..d3e31369a172 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java @@ -1130,11 +1130,14 @@ public static void validateFileContent(byte[] concat, byte[][] bytes) { * Utility to validate vectored read results. * @param fileRanges input ranges. * @param originalData original data. + * @param baseOffset base offset of the original data * @throws IOException any ioe. */ - public static void validateVectoredReadResult(List fileRanges, - byte[] originalData) - throws IOException, TimeoutException { + public static void validateVectoredReadResult( + final List fileRanges, + final byte[] originalData, + final long baseOffset) + throws IOException, TimeoutException { CompletableFuture[] completableFutures = new CompletableFuture[fileRanges.size()]; int i = 0; for (FileRange res : fileRanges) { @@ -1150,8 +1153,8 @@ public static void validateVectoredReadResult(List fileRanges, ByteBuffer buffer = FutureIO.awaitFuture(data, VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); - assertDatasetEquals((int) res.getOffset(), "vecRead", - buffer, res.getLength(), originalData); + assertDatasetEquals((int) (res.getOffset() - baseOffset), "vecRead", + buffer, res.getLength(), originalData); } } @@ -1758,6 +1761,44 @@ public static long readStream(InputStream in) { } } + /** + * Create a range list with a single range within it. + * @param offset offset + * @param length length + * @return the list. + */ + public static List range( + final long offset, + final int length) { + return range(new ArrayList<>(), offset, length); + } + + /** + * Create a range and add it to the supplied list. + * @param fileRanges list of ranges + * @param offset offset + * @param length length + * @return the list. + */ + public static List range( + final List fileRanges, + final long offset, + final int length) { + fileRanges.add(FileRange.createFileRange(offset, length)); + return fileRanges; + } + + /** + * Given a list of ranges, calculate the total size. + * @param fileRanges range list. + * @return total size of all reads. + */ + public static long totalReadSize(final List fileRanges) { + return fileRanges.stream() + .mapToLong(FileRange::getLength) + .sum(); + } + /** * Results of recursive directory creation/scan operations. */ diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java index e039e14e47d0..83cda4b22198 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java @@ -32,10 +32,15 @@ import java.net.URI; import java.nio.ByteBuffer; import java.nio.file.Files; +import java.util.ArrayList; +import java.util.List; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.RandomStringUtils; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileRange; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hdds.client.DefaultReplicationConfig; @@ -324,4 +329,104 @@ public void testSequenceFileReaderSyncEC() throws IOException { assertEquals(srcfile.length(), blockStart); in.close(); } + + @Test + public void testVectoredRead() throws Exception { + try (FSDataInputStream inputStream = fs.open(filePath)) { + // Create multiple file ranges to read + List ranges = new ArrayList<>(); + + // Read from different parts of the file + ranges.add(FileRange.createFileRange(0L, 100)); // First 100 bytes + ranges.add(FileRange.createFileRange(500L, 200)); // 200 bytes from offset 500 + ranges.add(FileRange.createFileRange(1024L, 512)); // 512 bytes from offset 1024 + ranges.add(FileRange.createFileRange(10L * 1024, 1024)); // 1KB from offset 10KB + + // Perform vectored read + inputStream.readVectored(ranges, ByteBuffer::allocate); + + // Wait for all reads to complete and validate + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + ByteBuffer buffer = result.get(30, TimeUnit.SECONDS); + + assertEquals(range.getLength(), buffer.remaining(), + "Buffer size mismatch for range at offset " + range.getOffset()); + + // Validate data by comparing with sequential read + byte[] vectoredData = new byte[buffer.remaining()]; + buffer.get(vectoredData); + + byte[] expectedData = new byte[range.getLength()]; + System.arraycopy(data, (int)range.getOffset(), expectedData, 0, range.getLength()); + + assertArrayEquals(expectedData, vectoredData, + "Data mismatch for range at offset " + range.getOffset()); + } + } + } + + @Test + public void testVectoredReadWithOverlappingRanges() throws Exception { + try (FSDataInputStream inputStream = fs.open(filePath)) { + // Create overlapping ranges + List ranges = new ArrayList<>(); + ranges.add(FileRange.createFileRange(100L, 500)); + ranges.add(FileRange.createFileRange(300L, 400)); // Overlaps with first range + ranges.add(FileRange.createFileRange(600L, 200)); + + // Overlapping ranges should throw IllegalArgumentException + assertThrows(IllegalArgumentException.class, () -> { + inputStream.readVectored(ranges, ByteBuffer::allocate); + }); + } + } + + @Test + public void testVectoredReadAcrossMultipleBlocks() throws Exception { + // This test reads ranges that span across different blocks in the file + try (FSDataInputStream inputStream = fs.open(filePath)) { + List ranges = new ArrayList<>(); + + // Assuming blocks are 1MB each, read ranges that cross block boundaries + ranges.add(FileRange.createFileRange(1024L * 1024 - 100, 200)); // Crosses first block boundary + ranges.add(FileRange.createFileRange(2L * 1024 * 1024 - 50, 100)); // Crosses second block boundary + + inputStream.readVectored(ranges, ByteBuffer::allocate); + + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + ByteBuffer buffer = result.get(30, TimeUnit.SECONDS); + + byte[] vectoredData = new byte[buffer.remaining()]; + buffer.get(vectoredData); + + byte[] expectedData = new byte[range.getLength()]; + System.arraycopy(data, (int)range.getOffset(), expectedData, 0, range.getLength()); + + assertArrayEquals(expectedData, vectoredData, + "Data mismatch for range crossing block boundary at offset " + range.getOffset()); + } + } + } + + @Test + public void testVectoredReadWithSmallRanges() throws Exception { + try (FSDataInputStream inputStream = fs.open(filePath)) { + // Test with many small ranges + List ranges = new ArrayList<>(); + for (int i = 0; i < 50; i++) { + ranges.add(FileRange.createFileRange((long) i * 100, 50)); + } + + inputStream.readVectored(ranges, ByteBuffer::allocate); + + // Validate all small reads + for (FileRange range : ranges) { + CompletableFuture result = range.getData(); + ByteBuffer buffer = result.get(30, TimeUnit.SECONDS); + assertEquals(range.getLength(), buffer.remaining()); + } + } + } } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/contract/AbstractOzoneContractTest.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/contract/AbstractOzoneContractTest.java index 7315abd1553a..95f50d2cbec4 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/contract/AbstractOzoneContractTest.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/contract/AbstractOzoneContractTest.java @@ -34,6 +34,7 @@ import org.apache.hadoop.fs.contract.AbstractContractRootDirectoryTest; import org.apache.hadoop.fs.contract.AbstractContractSeekTest; import org.apache.hadoop.fs.contract.AbstractContractUnbufferTest; +import org.apache.hadoop.fs.contract.AbstractContractVectoredReadTest; import org.apache.hadoop.fs.contract.AbstractFSContract; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.ozone.MiniOzoneCluster; @@ -43,6 +44,8 @@ import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.params.ParameterizedClass; +import org.junit.jupiter.params.provider.MethodSource; /** * Base class for Ozone contract tests. Manages lifecycle of {@link MiniOzoneCluster}. @@ -271,6 +274,25 @@ protected AbstractFSContract createContract(Configuration conf) { } } + @Nested + @ParameterizedClass(name = "buffer-{0}") + @MethodSource("params") + class TestContractVectoredRead extends AbstractContractVectoredReadTest { + TestContractVectoredRead(String bufferType) { + super(bufferType); + } + + @Override + protected Configuration createConfiguration() { + return createOzoneConfig(); + } + + @Override + protected AbstractFSContract createContract(Configuration conf) { + return createOzoneContract(conf); + } + } + @Nested class TestContractLeaseRecovery extends AbstractContractLeaseRecoveryTest { 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 e640c1e6d175..96a74679aaaa 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 @@ -22,16 +22,20 @@ import java.io.InputStream; import java.nio.ByteBuffer; import java.nio.ReadOnlyBufferException; +import java.util.List; +import java.util.function.IntFunction; import org.apache.hadoop.fs.ByteBufferPositionedReadable; import org.apache.hadoop.fs.ByteBufferReadable; import org.apache.hadoop.fs.CanUnbuffer; import org.apache.hadoop.fs.FSInputStream; +import org.apache.hadoop.fs.FileRange; import org.apache.hadoop.fs.FileSystem.Statistics; import org.apache.hadoop.fs.Seekable; import org.apache.hadoop.hdds.annotation.InterfaceAudience; import org.apache.hadoop.hdds.annotation.InterfaceStability; import org.apache.hadoop.hdds.scm.storage.ExtendedInputStream; import org.apache.hadoop.hdds.tracing.TracingUtil; +import org.apache.hadoop.hdds.utils.VectoredReadUtils; /** * The input stream for Ozone file system. @@ -165,7 +169,7 @@ public void unbuffer() { * @throws IOException if there is some error performing the read */ @Override - public int read(long position, ByteBuffer buf) throws IOException { + public synchronized int read(long position, ByteBuffer buf) throws IOException { if (!buf.hasRemaining()) { return 0; } @@ -197,7 +201,7 @@ public int read(long position, ByteBuffer buf) throws IOException { * @throws EOFException if end of file reached before reading fully */ @Override - public void readFully(long position, ByteBuffer buf) throws IOException { + public synchronized void readFully(long position, ByteBuffer buf) throws IOException { int bytesRead; for (int readCount = 0; buf.hasRemaining(); readCount += bytesRead) { bytesRead = this.read(position + (long)readCount, buf); @@ -207,4 +211,44 @@ public void readFully(long position, ByteBuffer buf) throws IOException { } } } + + /** + * Implements vectored read by reading each range asynchronously. + * This allows clients to read multiple byte ranges from the same file + * in a single call, potentially improving performance by enabling + * parallel reads and reducing round-trip overhead. + * + * @param ranges list of file ranges to read + * @param allocate function to allocate ByteBuffer for each range + * @throws IOException if there is an error performing the reads + * @apiNote This method is synchronized to prevent race conditions from + * concurrent readVectored() calls on the same stream instance. + */ + @Override + public synchronized void readVectored(List ranges, + IntFunction allocate) throws IOException { + TracingUtil.executeInNewSpan("OzoneFSInputStream.readVectored", () -> { + // Save the initial position + final long initialPosition = getPos(); + + // Use common vectored read implementation + VectoredReadUtils.performVectoredRead( + ranges, + allocate, + (offset, buffer) -> { + // readFully is synchronized and uses positioned reads + // which automatically preserve stream position + readFully(offset, buffer); + if (statistics != null) { + statistics.incrementBytesRead(buffer.remaining()); + } + } + ); + + // Restore position before returning from method + seek(initialPosition); + + return null; + }); + } }