Flink内存管理: 内存管理器

TaskManager内存模型

内存管理器作用

​ 我们有一块内存叫做管理内存. 内存管理器就是用来管理这块内存的.你设计的这块内存是不是得有人来操作维护的呀? 就是内存管理器维护管理内存的.

​

​ Flink 1.10 对TaskManager的内存模型和Flink应用程序的配置选项进行了重大更改，让用户能够更加严格地控制其内存开销。

    1.10之前版本，负责TaskManager所有内存。
    1.10版本开始，管理范围是Slot级别。

堆外内存资源申请

MemoryManager.java

public void allocatePages(
			Object owner,
			Collection<MemorySegment> target,
			int numberOfPages) throws MemoryAllocationException {
		... ...

		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;
		});
		... ...
}

MemorySegmentFactory.java

public static MemorySegment allocateOffHeapUnsafeMemory(int size, Object owner, Runnable customCleanupAction) {
	long address = MemoryUtils.allocateUnsafe(size);
	ByteBuffer offHeapBuffer = MemoryUtils.wrapUnsafeMemoryWithByteBuffer(address, size);
	MemoryUtils.createMemoryGcCleaner(offHeapBuffer, address, customCleanupAction);
	return new HybridMemorySegment(offHeapBuffer, owner);
}

RocksDB自己负责内存申请和释放

RocksDBOperationUtils.java

public static OpaqueMemoryResource<RocksDBSharedResources> allocateSharedCachesIfConfigured(
			RocksDBMemoryConfiguration memoryConfig,
			MemoryManager memoryManager,
			double memoryFraction,
			Logger logger) throws IOException {

	... ...

		try {
			if (memoryConfig.isUsingFixedMemoryPerSlot()) {
				assert memoryConfig.getFixedMemoryPerSlot() != null;

				logger.info("Getting fixed-size shared cache for RocksDB.");
				return memoryManager.getExternalSharedMemoryResource(
						FIXED_SLOT_MEMORY_RESOURCE_ID, allocator, memoryConfig.getFixedMemoryPerSlot().getBytes());
			}
			else {
				logger.info("Getting managed memory shared cache for RocksDB.");
				return memoryManager.getSharedMemoryResourceForManagedMemory(MANAGED_MEMORY_RESOURCE_ID, allocator, memoryFraction);
			}
		}
	... ...
}

MemoryManager.java

public <T extends AutoCloseable> OpaqueMemoryResource<T> getExternalSharedMemoryResource(
		String type,
		LongFunctionWithException<T, Exception> initializer,
		long numBytes) throws Exception {

	// This object identifies the lease in this request. It is used only to identify the release operation.
	// Using the object to represent the lease is a bit nicer safer than just using a reference counter.
	final Object leaseHolder = new Object();

	final SharedResources.ResourceAndSize<T> resource =
											//获取内存资源后,get 或者 分配
			sharedResources.getOrAllocateSharedResource(type, leaseHolder, initializer, numBytes);
	// 创建资源释放函数  : 这表示管理内存器管理的RocksDB.但是RocksDB内存的申请和释放是由RocksDB自己来的.只不过RocksDB受这个管理器的管理
	final ThrowingRunnable<Exception> disposer = () -> sharedResources.release(type, leaseHolder);

	return new OpaqueMemoryResource<>(resource.resourceHandle(), resource.size(), disposer);
}

赏

谢谢你请我吃糖果

支付宝

微信

• 本文作者： xubatian
• 本文链接： http://xubatian.cn/Flink内存管理-内存管理器/
• 版权声明： 本博客所有文章除特别声明外均为原创，采用 CC BY 4.0 CN协议 许可协议。转载请注明出处:https://www.xubatian.cn/