taklwu commented on code in PR #11102:
URL: https://github.com/apache/ozone/pull/11102#discussion_r3906124855
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -486,6 +487,106 @@ 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 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
+ */
+ int readPositioned(long blockRelativePosition, ByteBuffer dst)
+ throws IOException {
+ if (!initialized) {
+ initialize();
+ }
+ final long[] offsets = chunkOffsets;
+ final BlockData currentBlockData = blockData;
+ final long blockLength = length;
Review Comment:
sorry, may ask why it need to be synchronized? I thought `chunkOffsets`,
`blockData` and `length` were being initialized once in the `initialize()`
which is a `synchronized` function already, here the `readPositioned`
implemented as stateless and is trying to getting the the snapshot of this read
only information for further operation within `readPositioned`
maybe I missed something that these three data will be changed over time?
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java:
##########
@@ -486,6 +487,106 @@ 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 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
+ */
+ int readPositioned(long blockRelativePosition, ByteBuffer dst)
+ throws IOException {
+ if (!initialized) {
+ initialize();
+ }
+ final long[] offsets = chunkOffsets;
+ final BlockData currentBlockData = blockData;
+ final long blockLength = length;
+ if (offsets == null || currentBlockData == null
+ || blockRelativePosition < 0 || blockRelativePosition >= blockLength) {
+ return EOF;
+ }
+
+ final List<ChunkInfo> chunkInfos = currentBlockData.getChunksList();
+ int index = Arrays.binarySearch(offsets, blockRelativePosition);
+ if (index < 0) {
+ index = -index - 2;
+ }
+
+ long pos = blockRelativePosition;
+ 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;
+ }
+ final int numBytesRead =
+ readChunkAt(chunkInfo, chunkOffset, (int) numBytesToRead, dst);
+ totalReadLen += numBytesRead;
+ pos += numBytesRead;
+ index++;
+ }
+ return totalReadLen == 0 ? EOF : totalReadLen;
+ }
+
+ /**
+ * 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 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);
Review Comment:
I rechecked with below, so it seems the synchronized should not be required
in the body
**gRPC path (blockFileInputStream == null):**
- The ephemeral ChunkInputStream gets its own xceiverClient via
ChunkInputStream.acquireClient() (synchronized per instance).
- That uses xceiverClientFactory.acquireClientForReadData(), which is
thread-safe internally (XceiverClientManager synchronizes on its cache).
- Block-level xceiverClientGrpc / xceiverClientShortCircuit are only used
for getBlockData() during init, not for pread data reads.
**Short-circuit path:**
- LocalChunkInputStream uses positional FileChannel.read(buffer, pos) so
concurrent preads on the shared block channel are safe.
- Its acquireClient() is a no-op; it does not use the block’s short-circuit
client for reads
but I did checked that the real edge case is pread vs close()/unbuffer(),
which we can be fixed with a brief synchronized checkOpen() + snapshot before
createChunkInputStream, without synchronizing the whole pread
I will address it in next revision.
--
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]
