Flink 源码阅读笔记（7）- 内存管理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

public abstract class MemorySegment {
	protected final byte[] heapMemory; //堆内存引用
	protected long address; //堆外内存地址

	//基于堆内存创建MemorySegment
	MemorySegment(byte[] buffer, Object owner) {
		if (buffer == null) {
			throw new NullPointerException("buffer");
		}
		this.heapMemory = buffer;
		this.address = BYTE_ARRAY_BASE_OFFSET;
		this.size = buffer.length;
		this.addressLimit = this.address + this.size;
		this.owner = owner;
	}

	//基于堆外内存创建MemorySegment
	MemorySegment(long offHeapAddress, int size, Object owner) {
		if (offHeapAddress <= 0) {
			throw new IllegalArgumentException("negative pointer or size");
		}
		if (offHeapAddress >= Long.MAX_VALUE - Integer.MAX_VALUE) {
			// this is necessary to make sure the collapsed checks are safe against numeric overflows
			throw new IllegalArgumentException("Segment initialized with too large address: " + offHeapAddress
					+ " ; Max allowed address is " + (Long.MAX_VALUE - Integer.MAX_VALUE - 1));
		}
		this.heapMemory = null;
		this.address = offHeapAddress;
		this.addressLimit = this.address + size;
		this.size = size;
		this.owner = owner;
	}

	public boolean isOffHeap() {
		return heapMemory == null;
	}

	public final long getLong(int index) {
		final long pos = address + index;
		if (index >= 0 && pos <= addressLimit - 8) {
			//这是能够在一个实现中同时操作对内存和堆外内存的关键
			return UNSAFE.getLong(heapMemory, pos);
		}
		else if (address > addressLimit) {
			throw new IllegalStateException("segment has been freed");
		}
		else {
			// index is in fact invalid
			throw new IndexOutOfBoundsException();
		}
	}

	//......
}


public final class HybridMemorySegment extends MemorySegment {
	private final ByteBuffer offHeapBuffer;

	//堆外内存初始化
	HybridMemorySegment(ByteBuffer buffer, Object owner) {
		super(checkBufferAndGetAddress(buffer), buffer.capacity(), owner);
		this.offHeapBuffer = buffer;
	}

	//堆内内存初始化
	HybridMemorySegment(byte[] buffer, Object owner) {
		super(buffer, owner);
		this.offHeapBuffer = null;
	}

	//......
}

public final class HeapMemorySegment extends MemorySegment {
	private byte[] memory;

	HeapMemorySegment(byte[] memory, Object owner) {
		super(Objects.requireNonNull(memory), owner);
		this.memory = memory;
	}

	//......
}

1

sun.misc.Unsafe.getLong(Object reference, long offset)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

class NetworkBufferPool implements BufferPoolFactory {
	//所有可用的MemorySegment，阻塞队列
	private final ArrayBlockingQueue<MemorySegment> availableMemorySegments;

	public NetworkBufferPool(int numberOfSegmentsToAllocate, int segmentSize) {
		this.totalNumberOfMemorySegments = numberOfSegmentsToAllocate;
		this.memorySegmentSize = segmentSize;
		final long sizeInLong = (long) segmentSize;

		try {
			this.availableMemorySegments = new ArrayBlockingQueue<>(numberOfSegmentsToAllocate);
		}
		catch (OutOfMemoryError err) {
			throw new OutOfMemoryError("Could not allocate buffer queue of length "
					+ numberOfSegmentsToAllocate + " - " + err.getMessage());
		}
		try {
			for (int i = 0; i < numberOfSegmentsToAllocate; i++) {
				//NetworkBufferPool 使用的 MemorySegment 全是堆外内存
				availableMemorySegments.add(MemorySegmentFactory.allocateUnpooledOffHeapMemory(segmentSize, null));
			}
		}
		catch (OutOfMemoryError err) {
			......
		}
		.......
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

public class MemoryManager {
	//管理所有的 MemorySegment
	private final MemoryPool memoryPool;


	public MemoryManager(long memorySize, int numberOfSlots, int pageSize,
							MemoryType memoryType, boolean preAllocateMemory) {
		// sanity checks
		......
		this.memoryType = memoryType;
		this.memorySize = memorySize;
		this.numberOfSlots = numberOfSlots;

		// assign page size and bit utilities
		this.pageSize = pageSize;
		this.roundingMask = ~((long) (pageSize - 1));

		final long numPagesLong = memorySize / pageSize;
		if (numPagesLong > Integer.MAX_VALUE) {
			throw new IllegalArgumentException("The given number of memory bytes (" + memorySize
					+ ") corresponds to more than MAX_INT pages.");
		}
		//所有可用的 MemorySegment 数量
		this.totalNumPages = (int) numPagesLong;
		if (this.totalNumPages < 1) {
			throw new IllegalArgumentException("The given amount of memory amounted to less than one page.");
		}

		this.allocatedSegments = new HashMap<Object, Set<MemorySegment>>();
		this.isPreAllocated = preAllocateMemory;

		this.numNonAllocatedPages = preAllocateMemory ? 0 : this.totalNumPages;
		//是否需要预分配内存，Streaming 不会预分配
		final int memToAllocate = preAllocateMemory ? this.totalNumPages : 0;

		switch (memoryType) {
			case HEAP:
			//堆上
				this.memoryPool = new HybridHeapMemoryPool(memToAllocate, pageSize);
				break;
			case OFF_HEAP:
			//堆外
				if (!preAllocateMemory) {
					LOG.warn("It is advisable to set 'taskmanager.memory.preallocate' to true when" +
						" the memory type 'taskmanager.memory.off-heap' is set to true.");
				}
				this.memoryPool = new HybridOffHeapMemoryPool(memToAllocate, pageSize);
				break;
			default:
				throw new IllegalArgumentException("unrecognized memory type: " + memoryType);
		}
	}

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

	static final class HybridHeapMemoryPool extends MemoryPool {
		private final ArrayDeque<byte[]> availableMemory;

		HybridHeapMemoryPool(int numInitialSegments, int segmentSize) {
			this.availableMemory = new ArrayDeque<>(numInitialSegments);
			this.segmentSize = segmentSize;
			for (int i = 0; i < numInitialSegments; i++) {
				//堆上直接使用byte数组
				this.availableMemory.add(new byte[segmentSize]);
			}
		}
	}

	static final class HybridOffHeapMemoryPool extends MemoryPool {
		private final ArrayDeque<ByteBuffer> availableMemory;

		HybridOffHeapMemoryPool(int numInitialSegments, int segmentSize) {
			this.availableMemory = new ArrayDeque<>(numInitialSegments);
			this.segmentSize = segmentSize;
			//堆外使用 DirectByteBuffer
			for (int i = 0; i < numInitialSegments; i++) {
				this.availableMemory.add(ByteBuffer.allocateDirect(segmentSize));
			}
		}
	}
}