1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

public class TaskManagerServices {
    ...
    private final MemoryManager memoryManager;
    private final NetworkEnvironment networkEnvironment;
    ...
     public static TaskManagerServices fromConfiguration() {
        ...
        final NetworkEnvironment network = createNetworkEnvironment(taskManagerServicesConfiguration, maxJvmHeapMemory);
        final MemoryManager memoryManager = createMemoryManager(taskManagerServicesConfiguration, freeHeapMemoryWithDefrag, maxJvmHeapMemory);
        ...
    }
}

 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

public class MemoryManager {
    private final MemoryPool memoryPool;    //内存池
    private final MemoryType memoryType;    //内存类型，HEAP/OFF_HEAP
    private final boolean isPreAllocated;    //是否预先分配
    private final HashMap<Object, Set<MemorySegment>> allocatedSegments;    //已分配的内存片段

    public MemoryManager(...) {
        ...
        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);
        }
    }

    public void allocatePages(Object owner, List<MemorySegment> target, int numPages);
       public void release(MemorySegment segment);
    public void releaseAll(Object owner);
}

 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

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(int numInitialSegments, int segmentSize) {
   this.availableMemory = new ArrayDeque<>(numInitialSegments);
   this.segmentSize = segmentSize;

   for (int i = 0; i < numInitialSegments; i++) {
      // 使用 new 在 jvm 堆上分配内存
      this.availableMemory.add(new byte[segmentSize]);
   }
}

HybridOffHeapMemoryPool(int numInitialSegments, int segmentSize) {
    this.availableMemory = new ArrayDeque<>(numInitialSegments);
    this.segmentSize = segmentSize;

    for (int i = 0; i < numInitialSegments; i++) {
        // 使用 allocateDirect 直接堆外内存
        this.availableMemory.add(ByteBuffer.allocateDirect(segmentSize));
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

public final class MemorySegmentFactory {
    ...
    // 从池外(jvm堆中)分配内存段
    public static MemorySegment allocateUnpooledSegment(int size, Object owner) {
        return new HybridMemorySegment(new byte[size], owner);
    }
    // 从内存池 中 获取堆内存片段
    public static MemorySegment wrapPooledHeapMemory(byte[] memory, Object owner) {
        return new HybridMemorySegment(memory, owner);
    }
    // 从内存池中 获取 非堆内存片段
    public static MemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner) {
        return new HybridMemorySegment(memory, owner);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

public abstract class MemorySegment {
    protected final byte[] heapMemory;    //堆内存
    protected long address;    //字节数组对应的相对地址（若heapMemory为null，即可能为off-heap内存的绝对地址
    protected final long addressLimit;    //
    protected static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;    //用来对堆/非堆内存进行操作，是JVM的非安全的API
    protected static final long BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);    //二进制字节数组的起始索引，相对于字节数组对象
    private static final boolean LITTLE_ENDIAN = (ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN);    //字节序
    protected final int size;    //内存段的字节数
    private final Object owner;    //该内存段owner
    ...
 }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

//堆内存
public final class HeapMemorySegment extends MemorySegment {
  // 指向heapMemory的额外引用，用来如数组越界的检查
  private byte[] memory;
  // 只能初始化堆内存
  HeapMemorySegment(byte[] memory, Object owner) {
    super(Objects.requireNonNull(memory), owner);
    this.memory = memory;
  }
  //...
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

  //混合内存

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

public interface DataInputView extends DataInput {
  void skipBytesToRead(int numBytes) throws IOException;
  int read(byte[] b, int off, int len) throws IOException;    //读取数据到b中
  int read(byte[] b) throws IOException;
}

public interface DataOutputView extends DataOutput {
    void skipBytesToWrite(int numBytes) throws IOException;
    void write(DataInputView source, int numBytes) throws IOException; //
}