聊聊flink的MemoryPool

朴灿烈づ我的快乐病毒、 2022-03-07 11:46 180阅读 0赞

## 序 ##

本文主要研究一下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();
        }

## 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]);
                }
            }
    
            @Override
            MemorySegment allocateNewSegment(Object owner) {
                return MemorySegmentFactory.allocateUnpooledSegment(segmentSize, owner);
            }
    
            @Override
            MemorySegment requestSegmentFromPool(Object owner) {
                byte[] buf = availableMemory.remove();
                return  MemorySegmentFactory.wrapPooledHeapMemory(buf, owner);
            }
    
            @Override
            void 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());
                }
            }
    
            @Override
            protected int getNumberOfAvailableMemorySegments() {
                return availableMemory.size();
            }
    
            @Override
            void 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));
                }
            }
    
            @Override
            MemorySegment allocateNewSegment(Object owner) {
                return MemorySegmentFactory.allocateUnpooledOffHeapMemory(segmentSize, owner);
            }
    
            @Override
            MemorySegment requestSegmentFromPool(Object owner) {
                ByteBuffer buf = availableMemory.remove();
                return MemorySegmentFactory.wrapPooledOffHeapMemory(buf, owner);
            }
    
            @Override
            void 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());
                }
            }
    
            @Override
            protected int getNumberOfAvailableMemorySegments() {
                return availableMemory.size();
            }
    
            @Override
            void 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][]

[MemoryManager]: https://ci.apache.org/projects/flink/flink-docs-release-1.7/api/java/org/apache/flink/runtime/memory/MemoryManager.html