[5/5] flink git commit: [FLINK-7517][network] let NettyBufferPool extend PooledByteBufAllocator

[sr...@apache.org]

[FLINK-7517][network] let NettyBufferPool extend PooledByteBufAllocator

Previously, NettyBufferPool only wrapped PooledByteBufAllocator but then, any
allocated buffer's alloc() method was returning the wrapped
PooledByteBufAllocator which allowed heap buffers again. By extending the
PooledByteBufAllocator, we prevent this loop hole.

This also fixes the invariant that a copy of a buffer should have the same
allocator.

This closes #4594.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fcdd56e5
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fcdd56e5
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fcdd56e5

Branch: refs/heads/master
Commit: fcdd56e548ddd1bd7475970bdc5718b7b18d9803
Parents: 622daa4
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Wed Aug 23 12:04:28 2017 +0200
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Fri Jan 5 15:19:00 2018 +0100

----------------------------------------------------------------------
 .../io/network/netty/NettyBufferPool.java       | 144 ++++++-------------
 .../io/network/netty/NettyBufferPoolTest.java   |   2 +-
 2 files changed, 41 insertions(+), 105 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/fcdd56e5/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyBufferPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyBufferPool.java
index dfd3c32..6d2a6c8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyBufferPool.java
@@ -19,7 +19,6 @@
 package org.apache.flink.runtime.io.network.netty;
 
 import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
-import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBufAllocator;
 import org.apache.flink.shaded.netty4.io.netty.buffer.CompositeByteBuf;
 import org.apache.flink.shaded.netty4.io.netty.buffer.PooledByteBufAllocator;
 
@@ -33,17 +32,14 @@ import scala.Option;
 import static org.apache.flink.util.Preconditions.checkArgument;
 
 /**
- * Wrapper around Netty's {@link PooledByteBufAllocator} with strict control
+ * Extends around Netty's {@link PooledByteBufAllocator} with strict control
  * over the number of created arenas.
  */
-public class NettyBufferPool implements ByteBufAllocator {
+public class NettyBufferPool extends PooledByteBufAllocator {
 
 	private static final Logger LOG = LoggerFactory.getLogger(NettyBufferPool.class);
 
-	/** The wrapped buffer allocator. */
-	private final PooledByteBufAllocator alloc;
-
-	/** PoolArena<ByteBuffer>[] via Reflection. */
+	/** <tt>PoolArena&lt;ByteBuffer&gt;[]</tt> via Reflection. */
 	private final Object[] directArenas;
 
 	/** Configured number of arenas. */
@@ -52,6 +48,25 @@ public class NettyBufferPool implements ByteBufAllocator {
 	/** Configured chunk size for the arenas. */
 	private final int chunkSize;
 
+	/** We strictly prefer direct buffers and disallow heap allocations. */
+	private static final boolean PREFER_DIRECT = true;
+
+	/**
+	 * Arenas allocate chunks of pageSize << maxOrder bytes. With these defaults, this results in
+	 * chunks of 16 MB.
+	 *
+	 * @see #MAX_ORDER
+	 */
+	private static final int PAGE_SIZE = 8192;
+
+	/**
+	 * Arenas allocate chunks of pageSize << maxOrder bytes. With these defaults, this results in
+	 * chunks of 16 MB.
+	 *
+	 * @see #PAGE_SIZE
+	 */
+	private static final int MAX_ORDER = 11;
+
 	/**
 	 * Creates Netty's buffer pool with the specified number of direct arenas.
 	 *
@@ -59,44 +74,35 @@ public class NettyBufferPool implements ByteBufAllocator {
 	 *                       slots)
 	 */
 	public NettyBufferPool(int numberOfArenas) {
+		super(
+			PREFER_DIRECT,
+			// No heap arenas, please.
+			0,
+			// Number of direct arenas. Each arena allocates a chunk of 16 MB, i.e.
+			// we allocate numDirectArenas * 16 MB of direct memory. This can grow
+			// to multiple chunks per arena during runtime, but this should only
+			// happen with a large amount of connections per task manager. We
+			// control the memory allocations with low/high watermarks when writing
+			// to the TCP channels. Chunks are allocated lazily.
+			numberOfArenas,
+			PAGE_SIZE,
+			MAX_ORDER);
+
 		checkArgument(numberOfArenas >= 1, "Number of arenas");
 		this.numberOfArenas = numberOfArenas;
 
-		// We strictly prefer direct buffers and disallow heap allocations.
-		boolean preferDirect = true;
-
 		// Arenas allocate chunks of pageSize << maxOrder bytes. With these
 		// defaults, this results in chunks of 16 MB.
-		int pageSize = 8192;
-		int maxOrder = 11;
-
-		this.chunkSize = pageSize << maxOrder;
-
-		// Number of direct arenas. Each arena allocates a chunk of 16 MB, i.e.
-		// we allocate numDirectArenas * 16 MB of direct memory. This can grow
-		// to multiple chunks per arena during runtime, but this should only
-		// happen with a large amount of connections per task manager. We
-		// control the memory allocations with low/high watermarks when writing
-		// to the TCP channels. Chunks are allocated lazily.
-		int numDirectArenas = numberOfArenas;
-
-		// No heap arenas, please.
-		int numHeapArenas = 0;
 
-		this.alloc = new PooledByteBufAllocator(
-				preferDirect,
-				numHeapArenas,
-				numDirectArenas,
-				pageSize,
-				maxOrder);
+		this.chunkSize = PAGE_SIZE << MAX_ORDER;
 
 		Object[] allocDirectArenas = null;
 		try {
-			Field directArenasField = alloc.getClass()
+			Field directArenasField = PooledByteBufAllocator.class
 					.getDeclaredField("directArenas");
 			directArenasField.setAccessible(true);
 
-			allocDirectArenas = (Object[]) directArenasField.get(alloc);
+			allocDirectArenas = (Object[]) directArenasField.get(this);
 		} catch (Exception ignored) {
 			LOG.warn("Memory statistics not available");
 		} finally {
@@ -217,40 +223,10 @@ public class NettyBufferPool implements ByteBufAllocator {
 	}
 
 	// ------------------------------------------------------------------------
-	// Delegate calls to the allocated and prohibit heap buffer allocations
+	// Prohibit heap buffer allocations
 	// ------------------------------------------------------------------------
 
 	@Override
-	public ByteBuf buffer() {
-		return alloc.buffer();
-	}
-
-	@Override
-	public ByteBuf buffer(int initialCapacity) {
-		return alloc.buffer(initialCapacity);
-	}
-
-	@Override
-	public ByteBuf buffer(int initialCapacity, int maxCapacity) {
-		return alloc.buffer(initialCapacity, maxCapacity);
-	}
-
-	@Override
-	public ByteBuf ioBuffer() {
-		return alloc.ioBuffer();
-	}
-
-	@Override
-	public ByteBuf ioBuffer(int initialCapacity) {
-		return alloc.ioBuffer(initialCapacity);
-	}
-
-	@Override
-	public ByteBuf ioBuffer(int initialCapacity, int maxCapacity) {
-		return alloc.ioBuffer(initialCapacity, maxCapacity);
-	}
-
-	@Override
 	public ByteBuf heapBuffer() {
 		throw new UnsupportedOperationException("Heap buffer");
 	}
@@ -266,31 +242,6 @@ public class NettyBufferPool implements ByteBufAllocator {
 	}
 
 	@Override
-	public ByteBuf directBuffer() {
-		return alloc.directBuffer();
-	}
-
-	@Override
-	public ByteBuf directBuffer(int initialCapacity) {
-		return alloc.directBuffer(initialCapacity);
-	}
-
-	@Override
-	public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
-		return alloc.directBuffer(initialCapacity, maxCapacity);
-	}
-
-	@Override
-	public CompositeByteBuf compositeBuffer() {
-		return alloc.compositeBuffer();
-	}
-
-	@Override
-	public CompositeByteBuf compositeBuffer(int maxNumComponents) {
-		return alloc.compositeBuffer(maxNumComponents);
-	}
-
-	@Override
 	public CompositeByteBuf compositeHeapBuffer() {
 		throw new UnsupportedOperationException("Heap buffer");
 	}
@@ -299,19 +250,4 @@ public class NettyBufferPool implements ByteBufAllocator {
 	public CompositeByteBuf compositeHeapBuffer(int maxNumComponents) {
 		throw new UnsupportedOperationException("Heap buffer");
 	}
-
-	@Override
-	public CompositeByteBuf compositeDirectBuffer() {
-		return alloc.compositeDirectBuffer();
-	}
-
-	@Override
-	public CompositeByteBuf compositeDirectBuffer(int maxNumComponents) {
-		return alloc.compositeDirectBuffer(maxNumComponents);
-	}
-
-	@Override
-	public boolean isDirectBufferPooled() {
-		return alloc.isDirectBufferPooled();
-	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/fcdd56e5/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyBufferPoolTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyBufferPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyBufferPoolTest.java
index 956d54c..9897ba8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyBufferPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyBufferPoolTest.java
@@ -25,7 +25,7 @@ import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 /**
- * Tests for the {@link io.netty.buffer.PooledByteBufAllocator} wrapper.
+ * Tests for the {@link NettyBufferPool} wrapper.
  */
 public class NettyBufferPoolTest {