gaoyunhaii 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_r382382196
##########
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:
I think use a new boolean field might be better. We could still use a
placeholder buffer when the message is to be released to make BufferResponse
always have non-null buffer, but we will not rely on placeholder buffer to
decide the process.
----------------------------------------------------------------
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
