yandrey321 commented on code in PR #9402:
URL: https://github.com/apache/ozone/pull/9402#discussion_r2782869871
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java:
##########
@@ -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<? extends FileRange> ranges,
+ IntFunction<ByteBuffer> 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) {
Review Comment:
is there a reason for not allowing concurrent reads?
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java:
##########
@@ -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(
Review Comment:
does it need to be syncronized?
##########
hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java:
##########
@@ -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 {
Review Comment:
what is the reason for adding syncrhonized here?
##########
hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java:
##########
@@ -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 {
Review Comment:
what is the reason for adding syncrhonized here?
##########
hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java:
##########
@@ -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<? extends FileRange> ranges,
Review Comment:
does it need to be synchronized?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
