序

本文主要研究一下flink的MemoryPool

MemoryPool

flink-runtime_2.11-1.7.2-sources.jar!/org/apache/flink/runtime/memory/MemoryManager.java

	abstract static class MemoryPool {abstract int getNumberOfAvailableMemorySegments();abstract MemorySegment allocateNewSegment(Object owner);abstract MemorySegment requestSegmentFromPool(Object owner);abstract void returnSegmentToPool(MemorySegment segment);abstract void clear();}

• MemoryPool定義了getNumberOfAvailableMemorySegments、allocateNewSegment、requestSegmentFromPool、returnSegmentToPool、clear這幾個抽象方法；它有HybridHeapMemoryPool、HybridOffHeapMemoryPool這兩個子類

HybridHeapMemoryPool

flink-runtime_2.11-1.7.2-sources.jar!/org/apache/flink/runtime/memory/MemoryManager.java

	static final class HybridHeapMemoryPool extends MemoryPool {/** The collection of available memory segments. */private final ArrayDeque<byte[]> availableMemory;private final int segmentSize;HybridHeapMemoryPool(int numInitialSegments, int segmentSize) {this.availableMemory = new ArrayDeque<>(numInitialSegments);this.segmentSize = segmentSize;for (int i = 0; i < numInitialSegments; i++) {this.availableMemory.add(new byte[segmentSize]);}}@OverrideMemorySegment allocateNewSegment(Object owner) {return MemorySegmentFactory.allocateUnpooledSegment(segmentSize, owner);}@OverrideMemorySegment requestSegmentFromPool(Object owner) {byte[] buf = availableMemory.remove();return  MemorySegmentFactory.wrapPooledHeapMemory(buf, owner);}@Overridevoid returnSegmentToPool(MemorySegment segment) {if (segment.getClass() == HybridMemorySegment.class) {HybridMemorySegment heapSegment = (HybridMemorySegment) segment;availableMemory.add(heapSegment.getArray());heapSegment.free();}else {throw new IllegalArgumentException("Memory segment is not a " + HybridMemorySegment.class.getSimpleName());}}@Overrideprotected int getNumberOfAvailableMemorySegments() {return availableMemory.size();}@Overridevoid clear() {availableMemory.clear();}}

• HybridHeapMemoryPool繼承了MemoryPool，它使用的是jvm的heap內存；構造器接收numInitialSegments、segmentSize兩個參數用于初始化availableMemory這個ArrayDeque，該queue的元素類型為byte[]
• allocateNewSegment方法調用的是MemorySegmentFactory.allocateUnpooledSegment，用于分配unpooled memory；requestSegmentFromPool方法調用的是availableMemory.remove()，然后調用MemorySegmentFactory.wrapPooledHeapMemory包裝為MemorySegment，這個方法沒有判斷ArrayDeque的大小就直接remove，需要注意
• returnSegmentToPool方法只對HybridMemorySegment類型進行處理，首先將它的byte[]歸還到availableMemory，之后調用heapSegment.free()釋放；getNumberOfAvailableMemorySegments方法返回的是availableMemory.size()；clear方法調用的是availableMemory.clear()

HybridOffHeapMemoryPool

flink-runtime_2.11-1.7.2-sources.jar!/org/apache/flink/runtime/memory/MemoryManager.java

	static final class HybridOffHeapMemoryPool extends MemoryPool {/** The collection of available memory segments. */private final ArrayDeque<ByteBuffer> availableMemory;private final int segmentSize;HybridOffHeapMemoryPool(int numInitialSegments, int segmentSize) {this.availableMemory = new ArrayDeque<>(numInitialSegments);this.segmentSize = segmentSize;for (int i = 0; i < numInitialSegments; i++) {this.availableMemory.add(ByteBuffer.allocateDirect(segmentSize));}}@OverrideMemorySegment allocateNewSegment(Object owner) {return MemorySegmentFactory.allocateUnpooledOffHeapMemory(segmentSize, owner);}@OverrideMemorySegment requestSegmentFromPool(Object owner) {ByteBuffer buf = availableMemory.remove();return MemorySegmentFactory.wrapPooledOffHeapMemory(buf, owner);}@Overridevoid returnSegmentToPool(MemorySegment segment) {if (segment.getClass() == HybridMemorySegment.class) {HybridMemorySegment hybridSegment = (HybridMemorySegment) segment;ByteBuffer buf = hybridSegment.getOffHeapBuffer();availableMemory.add(buf);hybridSegment.free();}else {throw new IllegalArgumentException("Memory segment is not a " + HybridMemorySegment.class.getSimpleName());}}@Overrideprotected int getNumberOfAvailableMemorySegments() {return availableMemory.size();}@Overridevoid clear() {availableMemory.clear();}}

• HybridOffHeapMemoryPool繼承了MemoryPool，它使用的是OffHeap；構造器接收numInitialSegments、segmentSize兩個參數用于初始化availableMemory這個ArrayDeque，該queue的元素類型為ByteBuffer
• allocateNewSegment方法調用的是MemorySegmentFactory.allocateUnpooledOffHeapMemory，用于分配unpooled off-heap memory；requestSegmentFromPool方法調用的是availableMemory.remove()，然后調用MemorySegmentFactory.wrapPooledOffHeapMemory包裝為MemorySegment，這個方法沒有判斷ArrayDeque的大小就直接remove，需要注意
• returnSegmentToPool方法只對HybridMemorySegment類型進行處理，首先將它的ByteBuffer歸還到availableMemory，之后調用heapSegment.free()釋放；getNumberOfAvailableMemorySegments方法返回的是availableMemory.size()；clear方法調用的是availableMemory.clear()

小結

• MemoryPool定義了getNumberOfAvailableMemorySegments、allocateNewSegment、requestSegmentFromPool、returnSegmentToPool、clear這幾個抽象方法；它有HybridHeapMemoryPool、HybridOffHeapMemoryPool這兩個子類
• HybridHeapMemoryPool繼承了MemoryPool，它使用的是jvm的heap內存；構造器接收numInitialSegments、segmentSize兩個參數用于初始化availableMemory這個ArrayDeque，該queue的元素類型為byte[]；allocateNewSegment方法調用的是MemorySegmentFactory.allocateUnpooledSegment，用于分配unpooled memory；requestSegmentFromPool方法調用的是availableMemory.remove()，然后調用MemorySegmentFactory.wrapPooledHeapMemory包裝為MemorySegment，這個方法沒有判斷ArrayDeque的大小就直接remove，需要注意；returnSegmentToPool方法只對HybridMemorySegment類型進行處理，首先將它的byte[]歸還到availableMemory，之后調用heapSegment.free()釋放；getNumberOfAvailableMemorySegments方法返回的是availableMemory.size()；clear方法調用的是availableMemory.clear()
• HybridOffHeapMemoryPool繼承了MemoryPool，它使用的是OffHeap；構造器接收numInitialSegments、segmentSize兩個參數用于初始化availableMemory這個ArrayDeque，該queue的元素類型為ByteBuffer；allocateNewSegment方法調用的是MemorySegmentFactory.allocateUnpooledOffHeapMemory，用于分配unpooled off-heap memory；requestSegmentFromPool方法調用的是availableMemory.remove()，然后調用MemorySegmentFactory.wrapPooledOffHeapMemory包裝為MemorySegment，這個方法沒有判斷ArrayDeque的大小就直接remove，需要注意；returnSegmentToPool方法只對HybridMemorySegment類型進行處理，首先將它的ByteBuffer歸還到availableMemory，之后調用heapSegment.free()釋放；getNumberOfAvailableMemorySegments方法返回的是availableMemory.size()；clear方法調用的是availableMemory.clear()

doc

• MemoryManager