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_r377460964
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
##########
@@ -351,38 +352,74 @@ 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);
+ }
+ } else {
+ dataBuffer =
bufferAllocator.allocateUnPooledNetworkBuffer(0);
+ }
+
+ if (dataBuffer == null) {
+ dataBuffer =
bufferAllocator.getPlaceHolderBuffer();
+ }
- ByteBuf retainedSlice = buffer.readSlice(size).retain();
- return new BufferResponse(retainedSlice, isBuffer,
isCompressed, sequenceNumber, receiverId, backlog);
+ dataBuffer.setCompressed(isCompressed);
+
+ return new BufferResponse(
+ dataBuffer,
+ isBuffer,
+ isCompressed,
+ sequenceNumber,
+ receiverId,
+ backlog,
+ size);
+ }
+
+ static int getMessageHeaderLength(int lengthWithoutFrameHeader)
{
Review comment:
Actually this way is still hacky, the specific message should be aware of
its length, but it is still understood by the upper layer and take a round to
get it.
----------------------------------------------------------------
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
