[FLINK-10339][network] Use off-heap memory for SpillReadBufferPool

tongcheng-elong · Oct 1, 2018 · 2ab55df · 2ab55df
1 parent 6ba2a0b
commit 2ab55df
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/.../src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionView.java b/.../src/main/java/org/apache/flink/runtime/io/network/partition/SpilledSubpartitionView.java
@@ -36,6 +36,7 @@
 import javax.annotation.concurrent.GuardedBy;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -257,7 +258,8 @@ private static class SpillReadBufferPool implements BufferRecycler {
 
  synchronized (buffers) {
  for (int i = 0; i < numberOfBuffers; i++) {
- buffers.add(new NetworkBuffer(MemorySegmentFactory.allocateUnpooledSegment(memorySegmentSize), this));
+ ByteBuffer memory = ByteBuffer.allocateDirect(memorySegmentSize);
+ buffers.add(new NetworkBuffer(MemorySegmentFactory.wrapPooledOffHeapMemory(memory, null), this));
  }
  }
  }