You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2017/10/08 19:38:20 UTC
[10/10] flink git commit: [FLINK-7310] [core] always use the
HybridMemorySegment
[FLINK-7310] [core] always use the HybridMemorySegment
Since we'd like to use our own off-heap buffers for network communication, we
cannot use HeapMemorySegment anymore and need to rely on HybridMemorySegment.
We thus drop any code that loads the HeapMemorySegment (it is still available
if needed) in favour of the HybridMemorySegment which is able to work on both
heap and off-heap memory.
For the performance penalty of this change compared to using HeapMemorySegment
alone, see this interesting blob article (from 2015):
https://flink.apache.org/news/2015/09/16/off-heap-memory.html
This closes #4445
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e14caef3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e14caef3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e14caef3
Branch: refs/heads/master
Commit: e14caef3794d0acde4bc86f67eedd8d032a3f213
Parents: c3235c3
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Wed Aug 2 11:34:54 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 20:57:07 2017 +0200
----------------------------------------------------------------------
.../flink/core/memory/HeapMemorySegment.java | 36 +++--
.../flink/core/memory/HybridMemorySegment.java | 53 ++-----
.../flink/core/memory/MemorySegmentFactory.java | 144 ++-----------------
.../core/memory/MemorySegmentFactoryTest.java | 67 ---------
.../io/network/buffer/NetworkBufferPool.java | 2 +-
.../flink/runtime/memory/MemoryManager.java | 40 +++---
.../TaskManagerServicesConfiguration.java | 17 ---
.../PartitionRequestClientHandlerTest.java | 4 +-
.../operators/util/HashVsSortMiniBenchmark.java | 30 ++++
9 files changed, 99 insertions(+), 294 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java b/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
index 9aa81d0..96c013a 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
@@ -37,6 +37,7 @@ import java.util.Objects;
* <p>Note that memory segments should usually not be allocated manually, but rather through the
* {@link MemorySegmentFactory}.
*/
+@SuppressWarnings("unused")
@Internal
public final class HeapMemorySegment extends MemorySegment {
@@ -177,29 +178,44 @@ public final class HeapMemorySegment extends MemorySegment {
* A memory segment factory that produces heap memory segments. Note that this factory does not
* support to allocate off-heap memory.
*/
- public static final class HeapMemorySegmentFactory implements MemorySegmentFactory.Factory {
+ public static final class HeapMemorySegmentFactory {
- @Override
+ /**
+ * Creates a new memory segment that targets the given heap memory region.
+ *
+ * @param memory The heap memory region.
+ * @return A new memory segment that targets the given heap memory region.
+ */
public HeapMemorySegment wrap(byte[] memory) {
return new HeapMemorySegment(memory);
}
- @Override
+ /**
+ * Allocates some unpooled memory and creates a new memory segment that represents
+ * that memory.
+ *
+ * @param size The size of the memory segment to allocate.
+ * @param owner The owner to associate with the memory segment.
+ * @return A new memory segment, backed by unpooled heap memory.
+ */
public HeapMemorySegment allocateUnpooledSegment(int size, Object owner) {
return new HeapMemorySegment(new byte[size], owner);
}
- @Override
+ /**
+ * Creates a memory segment that wraps the given byte array.
+ *
+ * <p>This method is intended to be used for components which pool memory and create
+ * memory segments around long-lived memory regions.
+ *
+ * @param memory The heap memory to be represented by the memory segment.
+ * @param owner The owner to associate with the memory segment.
+ * @return A new memory segment representing the given heap memory.
+ */
public HeapMemorySegment wrapPooledHeapMemory(byte[] memory, Object owner) {
return new HeapMemorySegment(memory, owner);
}
- @Override
- public HeapMemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner) {
- throw new UnsupportedOperationException(
- "The MemorySegment factory was not initialized for off-heap memory.");
- }
-
/**
* Prevent external instantiation.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java b/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
index 0b23ef3..a2cf3d3 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
@@ -158,7 +158,7 @@ public final class HybridMemorySegment extends MemorySegment {
// ------------------------------------------------------------------------
@Override
- public byte get(int index) {
+ public final byte get(int index) {
final long pos = address + index;
if (index >= 0 && pos < addressLimit) {
return UNSAFE.getByte(heapMemory, pos);
@@ -173,7 +173,7 @@ public final class HybridMemorySegment extends MemorySegment {
}
@Override
- public void put(int index, byte b) {
+ public final void put(int index, byte b) {
final long pos = address + index;
if (index >= 0 && pos < addressLimit) {
UNSAFE.putByte(heapMemory, pos, b);
@@ -188,17 +188,17 @@ public final class HybridMemorySegment extends MemorySegment {
}
@Override
- public void get(int index, byte[] dst) {
+ public final void get(int index, byte[] dst) {
get(index, dst, 0, dst.length);
}
@Override
- public void put(int index, byte[] src) {
+ public final void put(int index, byte[] src) {
put(index, src, 0, src.length);
}
@Override
- public void get(int index, byte[] dst, int offset, int length) {
+ public final void get(int index, byte[] dst, int offset, int length) {
// check the byte array offset and length and the status
if ((offset | length | (offset + length) | (dst.length - (offset + length))) < 0) {
throw new IndexOutOfBoundsException();
@@ -219,7 +219,7 @@ public final class HybridMemorySegment extends MemorySegment {
}
@Override
- public void put(int index, byte[] src, int offset, int length) {
+ public final void put(int index, byte[] src, int offset, int length) {
// check the byte array offset and length
if ((offset | length | (offset + length) | (src.length - (offset + length))) < 0) {
throw new IndexOutOfBoundsException();
@@ -241,12 +241,12 @@ public final class HybridMemorySegment extends MemorySegment {
}
@Override
- public boolean getBoolean(int index) {
+ public final boolean getBoolean(int index) {
return get(index) != 0;
}
@Override
- public void putBoolean(int index, boolean value) {
+ public final void putBoolean(int index, boolean value) {
put(index, (byte) (value ? 1 : 0));
}
@@ -436,41 +436,4 @@ public final class HybridMemorySegment extends MemorySegment {
}
return getAddress(buffer);
}
-
- // -------------------------------------------------------------------------
- // Factoring
- // -------------------------------------------------------------------------
-
- /**
- * Base factory for hybrid memory segments.
- */
- public static final class HybridMemorySegmentFactory implements MemorySegmentFactory.Factory {
-
- @Override
- public HybridMemorySegment wrap(byte[] memory) {
- return new HybridMemorySegment(memory);
- }
-
- @Override
- public HybridMemorySegment allocateUnpooledSegment(int size, Object owner) {
- return new HybridMemorySegment(new byte[size], owner);
- }
-
- @Override
- public HybridMemorySegment wrapPooledHeapMemory(byte[] memory, Object owner) {
- return new HybridMemorySegment(memory, owner);
- }
-
- @Override
- public HybridMemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner) {
- return new HybridMemorySegment(memory, owner);
- }
-
- /**
- * Prevent external instantiation.
- */
- HybridMemorySegmentFactory() {}
- }
-
- public static final HybridMemorySegmentFactory FACTORY = new HybridMemorySegmentFactory();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
index 5411601..c235999 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
@@ -22,35 +22,27 @@ import org.apache.flink.annotation.Internal;
import java.nio.ByteBuffer;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
/**
- * A factory for memory segments. The purpose of this factory is to make sure that all memory segments
- * for heap data are of the same type. That way, the runtime does not mix the various specializations
- * of the {@link org.apache.flink.core.memory.MemorySegment}. Not mixing them has shown to be beneficial
- * to method specialization by the JIT and to overall performance.
+ * A factory for (hybrid) memory segments ({@link HybridMemorySegment}).
*
- * <p>Note that this factory auto-initialized to use {@link org.apache.flink.core.memory.HeapMemorySegment},
- * if a request to create a segment comes before the initialization.
+ * <p>The purpose of this factory is to make sure that all memory segments for heap data are of the
+ * same type. That way, the runtime does not mix the various specializations of the {@link
+ * MemorySegment}. Not mixing them has shown to be beneficial to method specialization by the JIT
+ * and to overall performance.
*/
@Internal
-public class MemorySegmentFactory {
-
- /**
- * The factory to use.
- */
- private static volatile Factory factory;
+public final class MemorySegmentFactory {
/**
* Creates a new memory segment that targets the given heap memory region.
- * This method should be used to turn short lived byte arrays into memory segments.
+ *
+ * <p>This method should be used to turn short lived byte arrays into memory segments.
*
* @param buffer The heap memory region.
* @return A new memory segment that targets the given heap memory region.
*/
public static MemorySegment wrap(byte[] buffer) {
- ensureInitialized();
- return factory.wrap(buffer);
+ return new HybridMemorySegment(buffer);
}
/**
@@ -79,8 +71,7 @@ public class MemorySegmentFactory {
* @return A new memory segment, backed by unpooled heap memory.
*/
public static MemorySegment allocateUnpooledSegment(int size, Object owner) {
- ensureInitialized();
- return factory.allocateUnpooledSegment(size, owner);
+ return new HybridMemorySegment(new byte[size], owner);
}
/**
@@ -94,8 +85,7 @@ public class MemorySegmentFactory {
* @return A new memory segment representing the given heap memory.
*/
public static MemorySegment wrapPooledHeapMemory(byte[] memory, Object owner) {
- ensureInitialized();
- return factory.wrapPooledHeapMemory(memory, owner);
+ return new HybridMemorySegment(memory, owner);
}
/**
@@ -110,117 +100,7 @@ public class MemorySegmentFactory {
* @return A new memory segment representing the given off-heap memory.
*/
public static MemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner) {
- ensureInitialized();
- return factory.wrapPooledOffHeapMemory(memory, owner);
- }
-
- // ------------------------------------------------------------------------
-
- /**
- * Initializes this factory with the given concrete factory, iff it is not yet initialized.
- * This also checks if the factory is already initialized to the exact same concrete factory
- * as given.
- *
- * @param f The concrete factory to use.
- * @return True, if the factory is initialized with the given concrete factory, or if it was already
- * initialized with the exact same concrete factory. False, if it is already initialized with
- * a different concrete factory.
- */
- public static boolean initializeIfNotInitialized(Factory f) {
- checkNotNull(f);
-
- synchronized (MemorySegmentFactory.class) {
- if (factory == null) {
- factory = f;
- return true;
- } else {
- return factory == f;
- }
- }
- }
-
- /**
- * Checks whether this memory segment factory has been initialized (with a type to produce).
- *
- * @return True, if the factory has been initialized, false otherwise.
- */
- public static boolean isInitialized() {
- return factory != null;
+ return new HybridMemorySegment(memory, owner);
}
- /**
- * Gets the factory. May return null, if the factory has not been initialized.
- *
- * @return The factory, or null, if the factory has not been initialized.
- */
- public static Factory getFactory() {
- return factory;
- }
-
- /**
- * Sets the factory to the {@link HeapMemorySegment#FACTORY} if no factory has been initialized, yet.
- */
- private static void ensureInitialized() {
- if (factory == null) {
- synchronized (MemorySegmentFactory.class) {
- if (factory == null) {
- factory = HeapMemorySegment.FACTORY;
- }
- }
- }
- }
-
- // ------------------------------------------------------------------------
- // Internal factory
- // ------------------------------------------------------------------------
-
- /**
- * A concrete factory for memory segments.
- */
- public interface Factory {
-
- /**
- * Creates a new memory segment that targets the given heap memory region.
- *
- * @param memory The heap memory region.
- * @return A new memory segment that targets the given heap memory region.
- */
- MemorySegment wrap(byte[] memory);
-
- /**
- * Allocates some unpooled memory and creates a new memory segment that represents
- * that memory.
- *
- * @param size The size of the memory segment to allocate.
- * @param owner The owner to associate with the memory segment.
- * @return A new memory segment, backed by unpooled heap memory.
- */
- MemorySegment allocateUnpooledSegment(int size, Object owner);
-
- /**
- * Creates a memory segment that wraps the given byte array.
- *
- * <p>This method is intended to be used for components which pool memory and create
- * memory segments around long-lived memory regions.
- *
- *
- * @param memory The heap memory to be represented by the memory segment.
- * @param owner The owner to associate with the memory segment.
- * @return A new memory segment representing the given heap memory.
- */
- MemorySegment wrapPooledHeapMemory(byte[] memory, Object owner);
-
- /**
- * Creates a memory segment that wraps the off-heap memory backing the given ByteBuffer.
- * Note that the ByteBuffer needs to be a <i>direct ByteBuffer</i>.
- *
- * <p>This method is intended to be used for components which pool memory and create
- * memory segments around long-lived memory regions.
- *
- * @param memory The byte buffer with the off-heap memory to be represented by the memory segment.
- * @param owner The owner to associate with the memory segment.
- * @return A new memory segment representing the given off-heap memory.
- */
- MemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner);
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java
deleted file mode 100644
index 3fd2cbe..0000000
--- a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.core.memory;
-
-import org.junit.Test;
-
-import java.lang.reflect.Field;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-/**
- * Tests for the {@link MemorySegmentFactory} in on/off-heap modes.
- */
-public class MemorySegmentFactoryTest {
-
- @Test
- public void testInitializationToHeapSegments() throws Exception {
- clearSegmentFactory();
-
- assertFalse(MemorySegmentFactory.isInitialized());
- assertTrue(MemorySegmentFactory.initializeIfNotInitialized(HeapMemorySegment.FACTORY));
- assertTrue(MemorySegmentFactory.isInitialized());
-
- assertTrue(MemorySegmentFactory.initializeIfNotInitialized(HeapMemorySegment.FACTORY));
- assertFalse(MemorySegmentFactory.initializeIfNotInitialized(HybridMemorySegment.FACTORY));
-
- assertEquals(HeapMemorySegment.FACTORY, MemorySegmentFactory.getFactory());
- }
-
- @Test
- public void testInitializationToOffHeapSegments() throws Exception {
- clearSegmentFactory();
-
- assertFalse(MemorySegmentFactory.isInitialized());
- assertTrue(MemorySegmentFactory.initializeIfNotInitialized(HybridMemorySegment.FACTORY));
- assertTrue(MemorySegmentFactory.isInitialized());
-
- assertTrue(MemorySegmentFactory.initializeIfNotInitialized(HybridMemorySegment.FACTORY));
- assertFalse(MemorySegmentFactory.initializeIfNotInitialized(HeapMemorySegment.FACTORY));
-
- assertEquals(HybridMemorySegment.FACTORY, MemorySegmentFactory.getFactory());
- }
-
- private static void clearSegmentFactory() throws Exception {
- Field factoryField = MemorySegmentFactory.class.getDeclaredField("factory");
- factoryField.setAccessible(true);
- factoryField.set(null, null);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
index 1eb44c3..b70e912 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
@@ -74,7 +74,7 @@ public class NetworkBufferPool implements BufferPoolFactory {
final long sizeInLong = (long) segmentSize;
try {
- this.availableMemorySegments = new ArrayBlockingQueue<MemorySegment>(numberOfSegmentsToAllocate);
+ this.availableMemorySegments = new ArrayBlockingQueue<>(numberOfSegmentsToAllocate);
}
catch (OutOfMemoryError err) {
throw new OutOfMemoryError("Could not allocate buffer queue of length "
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
index 192a586..c1a98cf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
@@ -18,9 +18,9 @@
package org.apache.flink.runtime.memory;
-import org.apache.flink.core.memory.HeapMemorySegment;
import org.apache.flink.core.memory.HybridMemorySegment;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.core.memory.MemoryType;
import org.apache.flink.util.MathUtils;
@@ -43,8 +43,8 @@ import java.util.Set;
* is represented in segments of equal size. Operators allocate the memory by requesting a number
* of memory segments.
*
- * <p>The memory may be represented as on-heap byte arrays ({@link HeapMemorySegment}), or as off-heap
- * memory regions ({@link HybridMemorySegment}). Which kind of memory the MemoryManager serves can
+ * <p>The memory may be represented as on-heap byte arrays or as off-heap memory regions
+ * (both via {@link HybridMemorySegment}). Which kind of memory the MemoryManager serves can
* be passed as an argument to the initialization.
*
* <p>The memory manager can either pre-allocate all memory, or allocate the memory on demand. In the
@@ -164,7 +164,7 @@ public class MemoryManager {
switch (memoryType) {
case HEAP:
- this.memoryPool = new HeapMemoryPool(memToAllocate, pageSize);
+ this.memoryPool = new HybridHeapMemoryPool(memToAllocate, pageSize);
break;
case OFF_HEAP:
if (!preAllocateMemory) {
@@ -598,15 +598,15 @@ public class MemoryManager {
abstract void clear();
}
- static final class HeapMemoryPool extends MemoryPool {
+ static final class HybridHeapMemoryPool extends MemoryPool {
/** The collection of available memory segments. */
private final ArrayDeque<byte[]> availableMemory;
private final int segmentSize;
- public HeapMemoryPool(int numInitialSegments, int segmentSize) {
- this.availableMemory = new ArrayDeque<byte[]>(numInitialSegments);
+ HybridHeapMemoryPool(int numInitialSegments, int segmentSize) {
+ this.availableMemory = new ArrayDeque<>(numInitialSegments);
this.segmentSize = segmentSize;
for (int i = 0; i < numInitialSegments; i++) {
@@ -615,25 +615,25 @@ public class MemoryManager {
}
@Override
- HeapMemorySegment allocateNewSegment(Object owner) {
- return HeapMemorySegment.FACTORY.allocateUnpooledSegment(segmentSize, owner);
+ MemorySegment allocateNewSegment(Object owner) {
+ return MemorySegmentFactory.allocateUnpooledSegment(segmentSize, owner);
}
@Override
- HeapMemorySegment requestSegmentFromPool(Object owner) {
+ MemorySegment requestSegmentFromPool(Object owner) {
byte[] buf = availableMemory.remove();
- return HeapMemorySegment.FACTORY.wrapPooledHeapMemory(buf, owner);
+ return MemorySegmentFactory.wrapPooledHeapMemory(buf, owner);
}
@Override
void returnSegmentToPool(MemorySegment segment) {
- if (segment.getClass() == HeapMemorySegment.class) {
- HeapMemorySegment heapSegment = (HeapMemorySegment) segment;
+ if (segment.getClass() == HybridMemorySegment.class) {
+ HybridMemorySegment heapSegment = (HybridMemorySegment) segment;
availableMemory.add(heapSegment.getArray());
heapSegment.free();
}
else {
- throw new IllegalArgumentException("Memory segment is not a " + HeapMemorySegment.class.getSimpleName());
+ throw new IllegalArgumentException("Memory segment is not a " + HybridMemorySegment.class.getSimpleName());
}
}
@@ -655,8 +655,8 @@ public class MemoryManager {
private final int segmentSize;
- public HybridOffHeapMemoryPool(int numInitialSegments, int segmentSize) {
- this.availableMemory = new ArrayDeque<ByteBuffer>(numInitialSegments);
+ HybridOffHeapMemoryPool(int numInitialSegments, int segmentSize) {
+ this.availableMemory = new ArrayDeque<>(numInitialSegments);
this.segmentSize = segmentSize;
for (int i = 0; i < numInitialSegments; i++) {
@@ -665,15 +665,15 @@ public class MemoryManager {
}
@Override
- HybridMemorySegment allocateNewSegment(Object owner) {
+ MemorySegment allocateNewSegment(Object owner) {
ByteBuffer memory = ByteBuffer.allocateDirect(segmentSize);
- return HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(memory, owner);
+ return MemorySegmentFactory.wrapPooledOffHeapMemory(memory, owner);
}
@Override
- HybridMemorySegment requestSegmentFromPool(Object owner) {
+ MemorySegment requestSegmentFromPool(Object owner) {
ByteBuffer buf = availableMemory.remove();
- return HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(buf, owner);
+ return MemorySegmentFactory.wrapPooledOffHeapMemory(buf, owner);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
index d1cfc4c..6cc7569 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
@@ -23,9 +23,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.QueryableStateOptions;
import org.apache.flink.configuration.TaskManagerOptions;
-import org.apache.flink.core.memory.HeapMemorySegment;
-import org.apache.flink.core.memory.HybridMemorySegment;
-import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.core.memory.MemoryType;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
@@ -281,20 +278,6 @@ public class TaskManagerServicesConfiguration {
memType = MemoryType.HEAP;
}
- // initialize the memory segment factory accordingly
- // TODO - this should be in the TaskManager, not the configuration
- if (memType == MemoryType.HEAP) {
- if (!MemorySegmentFactory.initializeIfNotInitialized(HeapMemorySegment.FACTORY)) {
- throw new Exception("Memory type is set to heap memory, but memory segment " +
- "factory has been initialized for off-heap memory segments");
- }
- } else {
- if (!MemorySegmentFactory.initializeIfNotInitialized(HybridMemorySegment.FACTORY)) {
- throw new Exception("Memory type is set to off-heap memory, but memory segment " +
- "factory has been initialized for heap memory segments");
- }
- }
-
// network buffer memory fraction
float networkBufFraction = configuration.getFloat(TaskManagerOptions.NETWORK_BUFFERS_MEMORY_FRACTION);
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
index ff19967..f3f6feb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
@@ -18,8 +18,8 @@
package org.apache.flink.runtime.io.network.netty;
-import org.apache.flink.core.memory.HeapMemorySegment;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferProvider;
import org.apache.flink.runtime.io.network.netty.NettyMessage.BufferResponse;
@@ -239,7 +239,7 @@ public class PartitionRequestClientHandlerTest {
// ---------------------------------------------------------------------------------------------
private static Buffer createBuffer(boolean fill) {
- MemorySegment segment = HeapMemorySegment.FACTORY.allocateUnpooledSegment(1024, null);
+ MemorySegment segment = MemorySegmentFactory.allocateUnpooledSegment(1024, null);
if (fill) {
for (int i = 0; i < 1024; i++) {
segment.put(i, (byte) i);
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/HashVsSortMiniBenchmark.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/HashVsSortMiniBenchmark.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/HashVsSortMiniBenchmark.java
index 4da2c8c..2696045 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/HashVsSortMiniBenchmark.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/HashVsSortMiniBenchmark.java
@@ -40,6 +40,7 @@ import org.apache.flink.runtime.operators.testutils.TestData.TupleGenerator.KeyM
import org.apache.flink.runtime.operators.testutils.TestData.TupleGenerator.ValueMode;
import org.apache.flink.util.Collector;
import org.apache.flink.util.MutableObjectIterator;
+
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -244,6 +245,35 @@ public class HashVsSortMiniBenchmark {
}
}
+ @Test
+ public void testSortOnly() throws Exception {
+ TestData.TupleGenerator generator1 = new TestData.TupleGenerator(SEED1, INPUT_1_SIZE / 10, 100, KeyMode.RANDOM, ValueMode.RANDOM_LENGTH);
+
+ final TestData.TupleGeneratorIterator input1 = new TestData.TupleGeneratorIterator(generator1, INPUT_1_SIZE);
+
+ long start = System.nanoTime();
+
+ final UnilateralSortMerger<Tuple2<Integer, String>> sorter = new UnilateralSortMerger<>(
+ this.memoryManager, this.ioManager, input1, this.parentTask, this.serializer1,
+ this.comparator1.duplicate(), (double)MEMORY_FOR_SORTER/MEMORY_SIZE, 128, 0.8f,
+ true /*use large record handler*/, true);
+
+ MutableObjectIterator<Tuple2<Integer, String>> iter = sorter.getIterator();
+
+ long stop1 = System.nanoTime();
+
+ Tuple2<Integer, String> t = new Tuple2<>();
+ while (iter.next() != null);
+
+ long stop2 = System.nanoTime();
+
+ long sortMsecs = (stop1 - start) / 1_000_000;
+ long allMsecs = (stop2 - start) / 1_000_000;
+
+ System.out.printf("Sort only took %d / %d msecs\n", sortMsecs, allMsecs);
+
+ sorter.close();
+ }
private static final class NoOpMatcher implements FlatJoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple2<Integer, String>> {
private static final long serialVersionUID = 1L;