Copilot commented on code in PR #11314:
URL: https://github.com/apache/ozone/pull/11314#discussion_r4091050453
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java:
##########
@@ -178,6 +181,139 @@ synchronized int readFully(ByteBuffer targetBuf, boolean
preRead) throws IOExcep
return read > 0 ? read : EOF;
}
+ /**
+ * Positioned read served by a dedicated one-shot streaming read which never
touches the
+ * cursor, the pre-read buffer or the sequential reader.
+ *
+ * @param blockOffset the offset within the block 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.
+ */
+ int readPositioned(long blockOffset, ByteBuffer dst) throws IOException {
+ if (!dst.hasRemaining()) {
+ return 0;
+ }
+ if (blockOffset < 0 || blockOffset >= blockLength) {
+ return EOF;
+ }
+ final XceiverClientFactory factory;
+ synchronized (this) {
+ checkOpen();
+ factory = xceiverClientFactory;
+ }
+ final int length = Math.toIntExact(Math.min(dst.remaining(), blockLength -
blockOffset));
+ final int startPosition = dst.position();
+ int callRetries = 0;
+ while (true) {
+ final AtomicReference<OneShotReader> readerRef = new AtomicReference<>();
+ try {
+ return preadOnce(factory, readerRef, blockOffset, length, dst);
+ } catch (IOException e) {
+ dst.position(startPosition);
+ handlePreadException(e, readerRef.get(), blockOffset, callRetries++);
+ }
+ }
+ }
+
+ private int preadOnce(XceiverClientFactory factory,
AtomicReference<OneShotReader> readerRef,
+ long blockOffset, int length, ByteBuffer dst) throws IOException {
+ final Pipeline pipeline = pipelineRef.get();
+ final XceiverClientSpi acquired =
factory.acquireClientForReadData(pipeline);
+ if (acquired == null) {
+ throw new IOException("Failed to acquire client for " + pipeline);
+ }
+ if (!(acquired instanceof XceiverClientGrpc)) {
+ throw new IOException("Unexpected client class: " +
acquired.getClass().getName() + ", " + pipeline);
+ }
+ final XceiverClientGrpc client = (XceiverClientGrpc) acquired;
+ try {
+ final OneShotReader reader = new OneShotReader(client);
+ readerRef.set(reader);
+ client.initStreamRead(blockID, reader, failedStreamingDatanodes);
+ try {
+ final StreamingReadResponse response = reader.getResponse();
+ if (response == null) {
+ throw new IOException("Uninitialized StreamingReadResponse: " +
blockID);
+ }
+ client.streamRead(ContainerProtocolCalls.buildReadBlockCommandProto(
+ blockID, blockOffset, length, responseDataSize, tokenRef.get(),
pipeline), response);
+
+ int copied = 0;
+ while (copied < length) {
+ final ReadBlockResponseProto proto = reader.poll();
+ if (proto == null) {
+ break;
+ }
+ final ByteBuffer buffer = getByteBuffer(proto, blockOffset + copied);
+ if (buffer == null || !buffer.hasRemaining()) {
+ continue;
+ }
+ final ByteBuffer tmpBuf = buffer.duplicate();
+ tmpBuf.limit(tmpBuf.position() + Math.min(buffer.remaining(), length
- copied));
+ copied += tmpBuf.remaining();
+ dst.put(tmpBuf);
+ }
+ return copied > 0 ? copied : EOF;
+ } finally {
+ closePreadReader(reader);
+ }
Review Comment:
If `initStreamRead` fails after trying its candidate datanodes,
`XceiverClientGrpc` has already released the semaphore permit for each failed
attempt. This `onCompleted()` still invokes `OneShotReader.releasePermit()` and
calls `completeStreamRead()` again, inflating the permit count; subsequent
reads can exceed the configured outstanding-stream limit. Track whether this
reader actually owns a permit before releasing it, while still closing any
request observer.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java:
##########
@@ -178,6 +181,139 @@ synchronized int readFully(ByteBuffer targetBuf, boolean
preRead) throws IOExcep
return read > 0 ? read : EOF;
}
+ /**
+ * Positioned read served by a dedicated one-shot streaming read which never
touches the
+ * cursor, the pre-read buffer or the sequential reader.
+ *
+ * @param blockOffset the offset within the block 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.
+ */
+ int readPositioned(long blockOffset, ByteBuffer dst) throws IOException {
+ if (!dst.hasRemaining()) {
+ return 0;
+ }
+ if (blockOffset < 0 || blockOffset >= blockLength) {
+ return EOF;
+ }
+ final XceiverClientFactory factory;
+ synchronized (this) {
+ checkOpen();
+ factory = xceiverClientFactory;
+ }
+ final int length = Math.toIntExact(Math.min(dst.remaining(), blockLength -
blockOffset));
+ final int startPosition = dst.position();
+ int callRetries = 0;
+ while (true) {
+ final AtomicReference<OneShotReader> readerRef = new AtomicReference<>();
+ try {
+ return preadOnce(factory, readerRef, blockOffset, length, dst);
+ } catch (IOException e) {
+ dst.position(startPosition);
+ handlePreadException(e, readerRef.get(), blockOffset, callRetries++);
+ }
+ }
+ }
+
+ private int preadOnce(XceiverClientFactory factory,
AtomicReference<OneShotReader> readerRef,
+ long blockOffset, int length, ByteBuffer dst) throws IOException {
+ final Pipeline pipeline = pipelineRef.get();
+ final XceiverClientSpi acquired =
factory.acquireClientForReadData(pipeline);
+ if (acquired == null) {
+ throw new IOException("Failed to acquire client for " + pipeline);
+ }
+ if (!(acquired instanceof XceiverClientGrpc)) {
+ throw new IOException("Unexpected client class: " +
acquired.getClass().getName() + ", " + pipeline);
+ }
Review Comment:
If the factory returns a non-gRPC client, this branch throws without
releasing the client acquired on line 221. The factory contract requires
callers to release acquired read clients, so an unexpected client type can leak
a pooled client; release `acquired` before throwing (or put the type check
inside the existing `try/finally`).
--
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]
