zhijiangW commented on a change in pull request #7368: [FLINK-10742][network]
Let Netty use Flink's buffers directly in credit-based mode
URL: https://github.com/apache/flink/pull/7368#discussion_r381779124
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
##########
@@ -351,38 +347,68 @@ ByteBuf write(ByteBufAllocator allocator) throws
IOException {
CompositeByteBuf composityBuf =
allocator.compositeDirectBuffer();
composityBuf.addComponent(headerBuf);
- composityBuf.addComponent(buffer);
+ composityBuf.addComponent(buffer.asByteBuf());
// update writer index since we have data
written to the components:
-
composityBuf.writerIndex(headerBuf.writerIndex() + buffer.writerIndex());
+
composityBuf.writerIndex(headerBuf.writerIndex() +
buffer.asByteBuf().writerIndex());
return composityBuf;
}
catch (Throwable t) {
if (headerBuf != null) {
headerBuf.release();
}
- buffer.release();
+ buffer.recycleBuffer();
ExceptionUtils.rethrowIOException(t);
return null; // silence the compiler
}
}
- static BufferResponse readFrom(ByteBuf buffer) {
- InputChannelID receiverId =
InputChannelID.fromByteBuf(buffer);
- int sequenceNumber = buffer.readInt();
- int backlog = buffer.readInt();
- boolean isBuffer = buffer.readBoolean();
- boolean isCompressed = buffer.readBoolean();
- int size = buffer.readInt();
+ /**
+ * Parses the message header part and composes a new
BufferResponse with an empty data buffer. The
+ * data buffer will be filled in later. This method is used in
credit-based network stack.
+ *
+ * @param messageHeader the serialized message header.
+ * @param bufferAllocator the allocator for network buffer.
+ * @return a BufferResponse object with the header parsed and
the data buffer to fill in later. The
+ * data buffer will be null if the target
channel has been released or the buffer size is 0.
+ */
+ static BufferResponse readFrom(ByteBuf messageHeader,
NetworkBufferAllocator bufferAllocator) {
+ InputChannelID receiverId =
InputChannelID.fromByteBuf(messageHeader);
+ int sequenceNumber = messageHeader.readInt();
+ int backlog = messageHeader.readInt();
+ boolean isBuffer = messageHeader.readBoolean();
+ boolean isCompressed = messageHeader.readBoolean();
+ int size = messageHeader.readInt();
+
+ Buffer dataBuffer = null;
+ if (size != 0) {
+ if (isBuffer) {
+ dataBuffer =
bufferAllocator.allocatePooledNetworkBuffer(receiverId);
+ } else {
+ dataBuffer =
bufferAllocator.allocateUnPooledNetworkBuffer(size);
+ }
+ }
+
+ if (dataBuffer == null) {
+ dataBuffer =
bufferAllocator.getPlaceHolderBuffer();
Review comment:
Based on the
[comment](https://github.com/apache/flink/pull/7368/files#r381770789), the
`BufferResponse` should provide a way for getting released info in order to be
handler by `PartitionRequestClientHandler`. There might be two options:
- Rely on whether `dataBuffer` is null, not quite sure whether it is
suggested to use `Optional` by the latest guideline .
- Define a new boolean field in `BufferResponse` to describe this info, but
it seems a bit redundant with `dataBuffer` state.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
