Copilot commented on code in PR #10743:
URL: https://github.com/apache/ozone/pull/10743#discussion_r3573642740
##########
pom.xml:
##########
@@ -188,7 +188,7 @@
<ratis-thirdparty.netty.version>4.1.130.Final</ratis-thirdparty.netty.version>
<ratis-thirdparty.protobuf.version>3.25.8</ratis-thirdparty.protobuf.version>
<ratis.thirdparty.version>1.0.11</ratis.thirdparty.version>
- <ratis.version>3.2.1</ratis.version>
+ <ratis.version>3.3.0-SNAPSHOT</ratis.version>
Review Comment:
Using <ratis.version>3.3.0-SNAPSHOT</ratis.version> makes the build
non-reproducible and (as noted in the PR description) currently fails in CI
unless a snapshot repository is configured. Consider keeping a released Ratis
version by default and gating the SNAPSHOT usage behind an explicit Maven
profile (or similar) so the default build remains green.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java:
##########
@@ -846,6 +860,222 @@ public CompletableFuture<Message> query(Message request) {
}
}
+ @Override
+ public void query(Message request, WritableByteChannel stream) {
+ try {
+ metrics.incNumQueryStateMachineOps();
+ final ContainerCommandRequestProto requestProto =
+ message2ContainerCommandRequestProto(request);
+ if (requestProto.getCmdType() == Type.ReadBlock) {
+ streamReadBlock(dispatcher, requestProto, stream);
+ } else {
+ writeAndClose(stream, dispatchCommand(requestProto, null));
+ }
+ } catch (IOException e) {
+ metrics.incNumQueryStateMachineFails();
+ throw new CompletionException(e);
+ }
+ }
+
+ static void streamReadBlock(
+ ContainerDispatcher dispatcher,
+ ContainerCommandRequestProto requestProto,
+ WritableByteChannel stream) throws IOException {
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+ final AtomicBoolean responseSeen = new AtomicBoolean(false);
+ final DataStreamObserver<ReadBlockResponse> observer =
+ new DataStreamObserver<ReadBlockResponse>() {
+ @Override
+ public void onNext(ReadBlockResponse response) {
+ responseSeen.set(true);
+ try {
+ writeReadBlockStreamResponse(stream, response.getResponse(),
+ response.getData());
+ } catch (IOException e) {
+ error.compareAndSet(null, e);
+ throw new CompletionException(e);
+ } catch (RuntimeException e) {
+ error.compareAndSet(null, e);
+ throw e;
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ error.set(throwable);
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+
+ };
+
+ try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+ dispatcher.streamDataReadOnly(requestProto, observer, blockFile, null);
+ } catch (CompletionException e) {
+ if (e.getCause() instanceof IOException) {
+ throw (IOException) e.getCause();
+ }
+ throw new IOException("Failed to stream ReadBlock " + requestProto,
+ e.getCause() != null ? e.getCause() : e);
+ }
+
+ if (error.get() != null) {
+ throw new IOException("Failed to stream ReadBlock " + requestProto,
+ error.get());
+ }
+ if (!responseSeen.get()) {
+ throw new IOException("ReadBlock stream returned no responses: "
+ + requestProto);
+ }
+
+ stream.close();
+ }
+
+ private static void writeAndClose(
+ WritableByteChannel stream, ContainerCommandResponseProto response)
+ throws IOException {
+ try {
+ writeFully(stream, response.toByteString().asReadOnlyByteBuffer());
+ } finally {
+ stream.close();
+ }
+ }
+
+ private static void writeReadBlockStreamResponse(
+ WritableByteChannel stream, ContainerCommandResponseProto response,
+ ByteBuffer data) throws IOException {
+ writeFully(stream, encodeReadBlockStreamResponse(response, data));
+ }
+
+ private static void writeFully(WritableByteChannel stream, ByteBuffer buffer)
+ throws IOException {
+ while (buffer.hasRemaining()) {
+ final int position = buffer.position();
+ final int remaining = buffer.remaining();
+ final int written = stream.write(buffer);
+ if (written < 0) {
+ throw new IOException("EOF reached while writing ReadBlock stream
response");
+ }
+ if (written == 0) {
+ throw new IOException("Zero bytes written to ReadBlock stream response
channel");
+ }
+ final int advanced = buffer.position() - position;
+ if (advanced == 0) {
+ if (written > remaining) {
+ throw new IOException("ReadBlock stream response write exceeded
remaining bytes: written="
+ + written + ", remaining=" + remaining);
+ }
+ buffer.position(position + written);
+ } else if (advanced != written) {
+ throw new IOException("Inconsistent ReadBlock stream response write:
written="
+ + written + ", advanced=" + advanced);
+ }
+ }
+ }
+
+ private static ByteBuffer encodeReadBlockStreamResponse(
+ ContainerCommandResponseProto response, ByteBuffer dataBuffer) {
+ final ByteBuffer data;
+ final ContainerCommandResponseProto metadata;
+ if (response.hasReadBlock()) {
+ final ReadBlockResponseProto readBlock = response.getReadBlock();
+ data = dataBuffer != null ? dataBuffer.asReadOnlyBuffer()
+ : readBlock.getData().asReadOnlyByteBuffer();
+ metadata = response.toBuilder()
+ .setReadBlock(readBlock.toBuilder().setData(ByteString.EMPTY))
+ .build();
+ } else {
+ data = ByteBuffer.allocate(0);
+ metadata = response;
+ }
+
+ final ByteBuffer metadataBuffer =
+ metadata.toByteString().asReadOnlyByteBuffer();
+ final ByteBuffer header =
+ ByteBuffer.allocate(RATIS_READ_BLOCK_STREAM_HEADER_BYTES);
+ header.putInt(metadataBuffer.remaining());
+ header.flip();
+ final ByteBuffer frame = ByteBuffer.allocate(
+ header.remaining() + metadataBuffer.remaining() + data.remaining());
+ frame.put(header);
+ frame.put(metadataBuffer);
+ frame.put(data);
Review Comment:
encodeReadBlockStreamResponse builds a single contiguous ByteBuffer frame by
copying metadata and the full data payload (frame.put(data)). This adds an
extra allocation and a full memcpy per chunk, undermining the goal of keeping
block data out of protobufs / minimizing copies for streaming reads. Prefer
writing header + metadata + data to the channel sequentially (or using a
GatheringByteChannel when available) to avoid copying the payload into an
intermediate buffer.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java:
##########
@@ -846,6 +860,222 @@ public CompletableFuture<Message> query(Message request) {
}
}
+ @Override
+ public void query(Message request, WritableByteChannel stream) {
+ try {
+ metrics.incNumQueryStateMachineOps();
+ final ContainerCommandRequestProto requestProto =
+ message2ContainerCommandRequestProto(request);
+ if (requestProto.getCmdType() == Type.ReadBlock) {
+ streamReadBlock(dispatcher, requestProto, stream);
+ } else {
+ writeAndClose(stream, dispatchCommand(requestProto, null));
+ }
+ } catch (IOException e) {
+ metrics.incNumQueryStateMachineFails();
+ throw new CompletionException(e);
+ }
+ }
+
+ static void streamReadBlock(
+ ContainerDispatcher dispatcher,
+ ContainerCommandRequestProto requestProto,
+ WritableByteChannel stream) throws IOException {
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+ final AtomicBoolean responseSeen = new AtomicBoolean(false);
+ final DataStreamObserver<ReadBlockResponse> observer =
+ new DataStreamObserver<ReadBlockResponse>() {
+ @Override
+ public void onNext(ReadBlockResponse response) {
+ responseSeen.set(true);
+ try {
+ writeReadBlockStreamResponse(stream, response.getResponse(),
+ response.getData());
+ } catch (IOException e) {
+ error.compareAndSet(null, e);
+ throw new CompletionException(e);
+ } catch (RuntimeException e) {
+ error.compareAndSet(null, e);
+ throw e;
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ error.set(throwable);
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+
+ };
+
+ try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+ dispatcher.streamDataReadOnly(requestProto, observer, blockFile, null);
+ } catch (CompletionException e) {
+ if (e.getCause() instanceof IOException) {
+ throw (IOException) e.getCause();
+ }
+ throw new IOException("Failed to stream ReadBlock " + requestProto,
+ e.getCause() != null ? e.getCause() : e);
+ }
+
+ if (error.get() != null) {
+ throw new IOException("Failed to stream ReadBlock " + requestProto,
+ error.get());
+ }
+ if (!responseSeen.get()) {
+ throw new IOException("ReadBlock stream returned no responses: "
+ + requestProto);
+ }
+
+ stream.close();
+ }
+
+ private static void writeAndClose(
+ WritableByteChannel stream, ContainerCommandResponseProto response)
+ throws IOException {
+ try {
+ writeFully(stream, response.toByteString().asReadOnlyByteBuffer());
+ } finally {
+ stream.close();
+ }
+ }
+
+ private static void writeReadBlockStreamResponse(
+ WritableByteChannel stream, ContainerCommandResponseProto response,
+ ByteBuffer data) throws IOException {
+ writeFully(stream, encodeReadBlockStreamResponse(response, data));
+ }
+
+ private static void writeFully(WritableByteChannel stream, ByteBuffer buffer)
+ throws IOException {
+ while (buffer.hasRemaining()) {
+ final int position = buffer.position();
+ final int remaining = buffer.remaining();
+ final int written = stream.write(buffer);
+ if (written < 0) {
+ throw new IOException("EOF reached while writing ReadBlock stream
response");
+ }
+ if (written == 0) {
+ throw new IOException("Zero bytes written to ReadBlock stream response
channel");
+ }
+ final int advanced = buffer.position() - position;
+ if (advanced == 0) {
+ if (written > remaining) {
+ throw new IOException("ReadBlock stream response write exceeded
remaining bytes: written="
+ + written + ", remaining=" + remaining);
+ }
+ buffer.position(position + written);
+ } else if (advanced != written) {
+ throw new IOException("Inconsistent ReadBlock stream response write:
written="
+ + written + ", advanced=" + advanced);
+ }
+ }
+ }
+
+ private static ByteBuffer encodeReadBlockStreamResponse(
+ ContainerCommandResponseProto response, ByteBuffer dataBuffer) {
+ final ByteBuffer data;
+ final ContainerCommandResponseProto metadata;
+ if (response.hasReadBlock()) {
+ final ReadBlockResponseProto readBlock = response.getReadBlock();
+ data = dataBuffer != null ? dataBuffer.asReadOnlyBuffer()
+ : readBlock.getData().asReadOnlyByteBuffer();
+ metadata = response.toBuilder()
+ .setReadBlock(readBlock.toBuilder().setData(ByteString.EMPTY))
+ .build();
+ } else {
+ data = ByteBuffer.allocate(0);
+ metadata = response;
+ }
+
+ final ByteBuffer metadataBuffer =
+ metadata.toByteString().asReadOnlyByteBuffer();
+ final ByteBuffer header =
+ ByteBuffer.allocate(RATIS_READ_BLOCK_STREAM_HEADER_BYTES);
+ header.putInt(metadataBuffer.remaining());
+ header.flip();
+ final ByteBuffer frame = ByteBuffer.allocate(
+ header.remaining() + metadataBuffer.remaining() + data.remaining());
+ frame.put(header);
+ frame.put(metadataBuffer);
+ frame.put(data);
+ frame.flip();
+ return frame.asReadOnlyBuffer();
+ }
+
+ private ContainerCommandResponseProto queryReadBlock(
+ ContainerCommandRequestProto requestProto) throws IOException {
+ final List<ContainerCommandResponseProto> responses = new ArrayList<>();
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+ final DataStreamObserver<ReadBlockResponse> observer =
+ new DataStreamObserver<ReadBlockResponse>() {
+ @Override
+ public void onNext(ReadBlockResponse response) {
+ responses.add(response.getData() == null ? response.getResponse()
+ : response.getResponse().toBuilder()
+
.setReadBlock(response.getResponse().getReadBlock().toBuilder()
+ .setData(ByteString.copyFrom(response.getData())))
+ .build());
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ error.set(throwable);
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ };
+
+ try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+ dispatcher.streamDataReadOnly(requestProto, observer, blockFile, null);
+ }
+
+ if (error.get() != null) {
+ throw new IOException("Failed to query ReadBlock " + requestProto,
+ error.get());
+ }
+ if (responses.isEmpty()) {
+ throw new IOException("ReadBlock query returned no responses: "
+ + requestProto);
+ }
+ return mergeReadBlockResponses(responses);
+ }
+
+ private static ContainerCommandResponseProto mergeReadBlockResponses(
+ List<ContainerCommandResponseProto> responses) {
+ final ContainerCommandResponseProto first = responses.get(0);
+ if (responses.size() == 1
+ || first.getResult() != ContainerProtos.Result.SUCCESS
+ || !first.hasReadBlock()) {
+ return first;
+ }
+
+ ByteString data = ByteString.EMPTY;
+ final ReadBlockResponseProto firstReadBlock = first.getReadBlock();
+ final ContainerProtos.ChecksumData.Builder checksum =
+ firstReadBlock.getChecksumData().toBuilder().clearChecksums();
+ for (ContainerCommandResponseProto response : responses) {
+ if (response.getResult() != ContainerProtos.Result.SUCCESS
+ || !response.hasReadBlock()) {
+ return response;
+ }
+ final ReadBlockResponseProto readBlock = response.getReadBlock();
+ data = data.concat(readBlock.getData());
+ checksum.addAllChecksums(readBlock.getChecksumData().getChecksumsList());
+ }
+
Review Comment:
mergeReadBlockResponses repeatedly concatenates ByteStrings in a loop (data
= data.concat(...)), which is O(n^2) in total bytes and can become expensive
for large blocks split into many responses. Accumulate into a ByteString.Output
(or similar) and convert once at the end to avoid repeated copying.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java:
##########
@@ -846,6 +860,222 @@ public CompletableFuture<Message> query(Message request) {
}
}
+ @Override
+ public void query(Message request, WritableByteChannel stream) {
+ try {
+ metrics.incNumQueryStateMachineOps();
+ final ContainerCommandRequestProto requestProto =
+ message2ContainerCommandRequestProto(request);
+ if (requestProto.getCmdType() == Type.ReadBlock) {
+ streamReadBlock(dispatcher, requestProto, stream);
+ } else {
+ writeAndClose(stream, dispatchCommand(requestProto, null));
+ }
+ } catch (IOException e) {
+ metrics.incNumQueryStateMachineFails();
+ throw new CompletionException(e);
+ }
+ }
+
+ static void streamReadBlock(
+ ContainerDispatcher dispatcher,
+ ContainerCommandRequestProto requestProto,
+ WritableByteChannel stream) throws IOException {
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+ final AtomicBoolean responseSeen = new AtomicBoolean(false);
+ final DataStreamObserver<ReadBlockResponse> observer =
+ new DataStreamObserver<ReadBlockResponse>() {
+ @Override
+ public void onNext(ReadBlockResponse response) {
+ responseSeen.set(true);
+ try {
+ writeReadBlockStreamResponse(stream, response.getResponse(),
+ response.getData());
+ } catch (IOException e) {
+ error.compareAndSet(null, e);
+ throw new CompletionException(e);
+ } catch (RuntimeException e) {
+ error.compareAndSet(null, e);
+ throw e;
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ error.set(throwable);
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+
+ };
+
+ try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+ dispatcher.streamDataReadOnly(requestProto, observer, blockFile, null);
+ } catch (CompletionException e) {
+ if (e.getCause() instanceof IOException) {
+ throw (IOException) e.getCause();
+ }
+ throw new IOException("Failed to stream ReadBlock " + requestProto,
+ e.getCause() != null ? e.getCause() : e);
+ }
+
+ if (error.get() != null) {
+ throw new IOException("Failed to stream ReadBlock " + requestProto,
+ error.get());
+ }
+ if (!responseSeen.get()) {
+ throw new IOException("ReadBlock stream returned no responses: "
+ + requestProto);
+ }
+
+ stream.close();
+ }
Review Comment:
streamReadBlock closes the WritableByteChannel only on the success path. If
dispatcher.streamDataReadOnly throws (or an exception is raised while writing a
response), the channel may be left open, which can leak resources / hang the
remote reader waiting for EOF. Wrap the streaming logic in a finally block to
ensure the channel is always closed (logging/suppressing close failures as
needed).
--
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]
