taklwu commented on code in PR #11102:
URL: https://github.com/apache/ozone/pull/11102#discussion_r3867024472
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -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<ChunkInputStream> 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);
Review Comment:
thanks, this is a important miss.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -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<ChunkInputStream> 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);
Review Comment:
thanks, this is an important miss.
--
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]
