Copilot commented on code in PR #10743:
URL: https://github.com/apache/ozone/pull/10743#discussion_r3573449844
##########
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));
+ }
Review Comment:
writeReadBlockStreamResponse currently allocates a single frame buffer and
copies the entire data payload into it (frame.put(data)), which defeats the
goal of streaming/zero-copy and adds extra heap pressure for large reads. Write
header/metadata and data as separate buffers instead.
##########
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 error/responseSeen checks throw, the
channel is leaked and the client may hang waiting for EOF.
##########
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");
+ }
Review Comment:
writeFully throws when WritableByteChannel.write returns 0, but a zero-byte
write is permitted (eg non-blocking channels or backpressure). Throwing here
can cause spurious failures under load; it should retry/yield instead.
--
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]
