文章目录
- 一.ManagedMemory（算子）内存的申请与使用
- 1. tm内存申请与使用大致流程
- 2. 创建MemoryManager实例
- 3. 算子使用通过MemoryManager使用内存
- 4. ManagedMemory内存空间申请流程
- 二.NetworkBuffer内存申请与使用
- 1. NetworkBuffer构造器
在Flink内存模型中我们已经知道，Flink会将内存按照使用方式、内存类型分为不同的内存区域，底层会借助MemorySegment对内存块进行管理和访问，MemorySegment的使用场景有很多，本文我们主要看下ManagedMemory和NetworkBuffer是如何申请和使用MemorySegment内存块的。
一.ManagedMemory（算子）内存的申请与使用
1. tm内存申请与使用大致流程
Task使用的物理计算资源主要是TaskSlot提供的，TaskSlot由TaskManager中TaskSlotTable组件创建和管理。
- 创建MemoryManager：JobManager申请到足够的Slot计算资源后，会在TaskSlotTable中创建相应的TaskSlot，然后对TaskSlot基本环境进行初始化，包括在TaskSlot内部创建MemoryManager组件。最终使用MemoryManager管理当前TaskSlot的内存计算资源。
- task线程使用内存：当Task线程启动时，会直接从TaskSlot中获取MemoryManager组件申请内存空间。通过MemoryManager对MemorySegment内存空间进行管理，这一步对应内存模型中的ManagedMemory，也被称为托管内存。
2. 创建MemoryManager实例
在TaskSlot的构造器中调用createMemoryManager()方法创建MemoryManager实例，管理当前TaskSlot(代表一个线程的资源) 中的内存空间
/**
创建具有**给定**容量和给定页面大小的内存管理器。
这是 MemoryManager 的生产版本，一旦 MemoryManager
的所有者准备好处置，它就会检查内存泄漏 ( verifyEmpty() )。
参数：
memorySize – 该内存管理器管理的堆外内存的总大小。
pageSize – 内存管理器分配的页面大小。
**/
private static MemoryManager createMemoryManager(
ResourceProfile resourceProfile, int pageSize) {
return MemoryManager.create(resourceProfile.getManagedMemory().getBytes(), pageSize);
}
在TaskSlot.createMemoryManager()方法中，会根据ResourceProfile参数获取内存空间大小，默认设置为非堆ing。其中pageSize参数就是MemorySegment的大小，如下代码默认为32kb。
TaskManagerOptions.
@Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER)
public static final ConfigOption<MemorySize> MEMORY_SEGMENT_SIZE =
key("taskmanager.memory.segment-size")
.memoryType()
.defaultValue(MemorySize.parse("32kb"))
.withDescription(
"Size of memory buffers used by the network stack and the memory manager.");
3. 算子使用通过MemoryManager使用内存
MemoryManager创建完毕后，会通过TaskSlot将MemoryManager对象传递给Task，此时Task会通过将MemoryManager封装在Environment变量中，然后传递给算子。
算子接收到MemoryManager对象后，通过MemoryManager动态申请内存空间，最终用于算子的具体计算过程。
需要注意的是：并不是所有的算子都会使用MemoryManager申请内存空间，这个步骤主要针对批计算类型的算子，例如HashJoinOperator、SortMergeJoinOperator和SortOperator等，这些算子往往需要借助非常大的内存空间进行数据的排序等操作。
4. ManagedMemory内存空间申请流程
申请ManagedMemory内存空间，是调用MemoryManager.allocatePages()方法执行的，见如下逻辑。
- 1）从AllocationRequest参数中获取MemorySegment的空集合、申请Pages总数量以及资源Owner（与内存关联的所有者：slot？还是算子？）等参数，并对参数进行非空和状态检查；
- 2）计算申请内存大小，并预留出内存空间；
- 3）根据page数、pageCleanup、owner等，开始分配内存，将内存以MemorySegment为单位，并维护一个set集合，最终返回给算子使用。
/**
从此内存管理器分配一组内存段。
分配的总内存不会超过构造函数中声明的大小限制。
参数：
owner – 与内存段关联的所有者，用于后备释放。
target – 将分配的内存页放入其中的列表。 numberOfPages – 要分配的页数。
**/
public void allocatePages(Object owner, Collection<MemorySegment> target, int numberOfPages)
throws MemoryAllocationException {
// sanity check
Preconditions.checkNotNull(owner, "The memory owner must not be null.");
Preconditions.checkState(!isShutDown, "Memory manager has been shut down.");
Preconditions.checkArgument(
numberOfPages <= totalNumberOfPages,
"Cannot allocate more segments %s than the max number %s",
numberOfPages,
totalNumberOfPages);
// reserve array space, if applicable
if (target instanceof ArrayList) {
((ArrayList<MemorySegment>) target).ensureCapacity(numberOfPages);
}
//计算申请内存大小，并预留空间（以免申请过程中被用掉）
long memoryToReserve = numberOfPages * pageSize;
try {
memoryBudget.reserveMemory(memoryToReserve);
} catch (MemoryReservationException e) {
throw new MemoryAllocationException(
String.format("Could not allocate %d pages", numberOfPages), e);
}
//创建pageCleanup方法用于清理unsafe内存
Runnable pageCleanup = this::releasePage;
allocatedSegments.compute(
owner,
(o, currentSegmentsForOwner) -> {
Set<MemorySegment> segmentsForOwner =
currentSegmentsForOwner == null
? new HashSet<>(numberOfPages)
: currentSegmentsForOwner;
for (long i = numberOfPages; i > 0; i--) {
//分配内存
MemorySegment segment =
allocateOffHeapUnsafeMemory(getPageSize(), owner, pageCleanup);
target.add(segment);
segmentsForOwner.add(segment);
}
return segmentsForOwner;
});
Preconditions.checkState(!isShutDown, "Memory manager has been concurrently shut down.");
}
如下如下算子会申请内存使用：
二.NetworkBuffer内存申请与使用
在Flink内存模型中，另外一个非常重要的堆外内存使用区域就是Network内存。Network内存主要用于网络传输中Buffer数据的缓冲区。
1. NetworkBuffer构造器
在NetworkBufferPool的构造器中可以看出，创建NetworkBufferPool时会根据用户配置的NetworkBuffer数量，调用MemorySegmentFactory创建相应的MemorySegment内存空间，再通过LocalBufferPool应用到ResultSubPartition或InputChannel组件中。
public NetworkBufferPool(
int numberOfSegmentsToAllocate, int segmentSize, Duration requestSegmentsTimeout) {
this.totalNumberOfMemorySegments = numberOfSegmentsToAllocate;
this.memorySegmentSize = segmentSize;
Preconditions.checkNotNull(requestSegmentsTimeout);
checkArgument(
requestSegmentsTimeout.toMillis() > 0,
"The timeout for requesting exclusive buffers should be positive.");
this.requestSegmentsTimeout = requestSegmentsTimeout;
final long sizeInLong = (long) segmentSize;
try {
this.availableMemorySegments = new ArrayDeque<>(numberOfSegmentsToAllocate);
} catch (OutOfMemoryError err) {
throw new OutOfMemoryError(
"Could not allocate buffer queue of length "
+ numberOfSegmentsToAllocate
+ " - "
+ err.getMessage());
}
try {
//申请segment内存，并放到availableMemorySegments中。
for (int i = 0; i < numberOfSegmentsToAllocate; i++) {
availableMemorySegments.add(
MemorySegmentFactory.allocateUnpooledOffHeapMemory(segmentSize, null));
}
} catch (OutOfMemoryError err) {
//如果申请过程中失败，则释放已申请的内存，算出缺少多少内存
int allocated = availableMemorySegments.size();
// free some memory
availableMemorySegments.clear();
long requiredMb = (sizeInLong * numberOfSegmentsToAllocate) >> 20;
long allocatedMb = (sizeInLong * allocated) >> 20;
long missingMb = requiredMb - allocatedMb;
throw new OutOfMemoryError(
"Could not allocate enough memory segments for NetworkBufferPool "
+ "(required (MB): "
+ requiredMb
+ ", allocated (MB): "
+ allocatedMb
+ ", missing (MB): "
+ missingMb
+ "). Cause: "
+ err.getMessage());
}
availabilityHelper.resetAvailable();
//计算共申请了多少mb：20：为2的20次方
long allocatedMb = (sizeInLong * availableMemorySegments.size()) >> 20;
LOG.info(
"Allocated {} MB for network buffer pool (number of memory segments: {}, bytes per segment: {}).",
allocatedMb,
availableMemorySegments.size(),
segmentSize);
}
参考：《Flink设计与实现：核心原理与源码解析》- 张利兵
