You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2017/10/04 13:14:08 UTC
flink git commit: [FLINK-7412][network] optimise
NettyMessage.TaskEventRequest#readFrom() to read from netty buffers directly
Repository: flink
Updated Branches:
refs/heads/master 82f27f5f1 -> 98ddbfd8e
[FLINK-7412][network] optimise NettyMessage.TaskEventRequest#readFrom() to read from netty buffers directly
This closes #4518.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/98ddbfd8
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/98ddbfd8
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/98ddbfd8
Branch: refs/heads/master
Commit: 98ddbfd8e6b6b38f1d6e61b5652d942650b214e7
Parents: 82f27f5
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Mon Aug 7 18:12:28 2017 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Oct 4 15:10:59 2017 +0200
----------------------------------------------------------------------
.../apache/flink/runtime/io/network/netty/NettyMessage.java | 9 ++++-----
1 file changed, 4 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/98ddbfd8/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
index 4989f03..d7ddfa6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
@@ -450,12 +450,11 @@ abstract class NettyMessage {
}
static TaskEventRequest readFrom(ByteBuf buffer, ClassLoader classLoader) throws IOException {
- // TODO Directly deserialize fromNetty's buffer
+ // directly deserialize fromNetty's buffer
int length = buffer.readInt();
- ByteBuffer serializedEvent = ByteBuffer.allocate(length);
-
- buffer.readBytes(serializedEvent);
- serializedEvent.flip();
+ ByteBuffer serializedEvent = buffer.nioBuffer(buffer.readerIndex(), length);
+ // assume this event's content is read from the ByteBuf (positions are not shared!)
+ buffer.readerIndex(buffer.readerIndex() + length);
TaskEvent event =
(TaskEvent) EventSerializer.fromSerializedEvent(serializedEvent, classLoader);