flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [11/15] flink git commit: [FLINK-1320] [core] Add an off-heap variant of the managed memory
Date Tue, 08 Sep 2015 18:59:01 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/MemorySegmentSpeedBenchmark.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/MemorySegmentSpeedBenchmark.java b/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/MemorySegmentSpeedBenchmark.java
new file mode 100644
index 0000000..454c821
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/MemorySegmentSpeedBenchmark.java
@@ -0,0 +1,1633 @@
+/*
+ * 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.benchmarks;
+
+import org.apache.flink.core.memory.HeapMemorySegment;
+import org.apache.flink.core.memory.HybridMemorySegment;
+import org.apache.flink.core.memory.MemorySegment;
+
+import java.nio.ByteBuffer;
+import java.util.Random;
+
+@SuppressWarnings("ConstantConditions")
+public class MemorySegmentSpeedBenchmark {
+	
+	private static final long LONG_VALUE = 0x1234567890abcdefl;
+	
+	private static final boolean TEST_CORE_ON_HEAP = true;
+	private static final boolean TEST_CORE_OFF_HEAP = false;
+	
+	// we keep this to make sure the JIT does not eliminate certain loops
+	public static long sideEffect = 0L;
+	
+	
+	public static void main(String[] args) {
+		
+		final int SMALL_SEGMENT_SIZE = 32 * 1024;
+		final int LARGE_SEGMENT_SIZE = 1024 * 1024 * 1024;
+		
+		final int SMALL_SEGMENTS_ROUNDS = 100000;
+		final int LARGE_SEGMENT_ROUNDS = 10;
+		
+		final byte[] largeSegment = new byte[LARGE_SEGMENT_SIZE];
+		final byte[] smallSegment = new byte[SMALL_SEGMENT_SIZE];
+		
+		final ByteBuffer largeOffHeap = ByteBuffer.allocateDirect(LARGE_SEGMENT_SIZE);
+		final ByteBuffer smallOffHeap = ByteBuffer.allocateDirect(SMALL_SEGMENT_SIZE);
+
+		System.out.println("testing access of individual bytes");
+		
+		testPutBytes(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE, SMALL_SEGMENTS_ROUNDS);
+		testGetBytes(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE, SMALL_SEGMENTS_ROUNDS);
+		testPutBytes(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE, LARGE_SEGMENT_ROUNDS);
+		testGetBytes(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE, LARGE_SEGMENT_ROUNDS);
+
+		System.out.println("testing access of byte arrays");
+
+		testPutByteArrays1024(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE / 1024, SMALL_SEGMENTS_ROUNDS);
+		testGetByteArrays1024(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE / 1024, SMALL_SEGMENTS_ROUNDS);
+		testPutByteArrays1024(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE / 1024, LARGE_SEGMENT_ROUNDS);
+		testGetByteArrays1024(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE / 1024, LARGE_SEGMENT_ROUNDS);
+		
+		System.out.println("testing access of longs");
+		
+		testPutLongs(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE / 8, SMALL_SEGMENTS_ROUNDS);
+		testGetLongs(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE / 8, SMALL_SEGMENTS_ROUNDS);
+		testPutLongs(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE / 8, LARGE_SEGMENT_ROUNDS);
+		testGetLongs(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE / 8, LARGE_SEGMENT_ROUNDS);
+
+//		System.out.println("testing access of big endian longs");
+//		
+//		testPutLongsBigEndian(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE / 8, SMALL_SEGMENTS_ROUNDS);
+//		testGetLongsBigEndian(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE / 8, SMALL_SEGMENTS_ROUNDS);
+//		testPutLongsBigEndian(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE / 8, LARGE_SEGMENT_ROUNDS);
+//		testGetLongsBigEndian(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE / 8, LARGE_SEGMENT_ROUNDS);
+//
+//		System.out.println("testing access of little endian longs");
+//		
+//		testPutLongsLittleEndian(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE / 8, SMALL_SEGMENTS_ROUNDS);
+//		testGetLongsLittleEndian(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE / 8, SMALL_SEGMENTS_ROUNDS);
+//		testPutLongsLittleEndian(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE / 8, LARGE_SEGMENT_ROUNDS);
+//		testGetLongsLittleEndian(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE / 8, LARGE_SEGMENT_ROUNDS);
+
+		System.out.println("testing access of ints");
+		
+		testPutInts(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE / 4, SMALL_SEGMENTS_ROUNDS);
+		testGetInts(smallSegment, smallOffHeap, SMALL_SEGMENT_SIZE / 4, SMALL_SEGMENTS_ROUNDS);
+		testPutInts(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE / 4, LARGE_SEGMENT_ROUNDS);
+		testGetInts(largeSegment, largeOffHeap, LARGE_SEGMENT_SIZE / 4, LARGE_SEGMENT_ROUNDS);
+
+
+	}
+
+	// --------------------------------------------------------------------------------------------
+	//                                  BYTEs
+	// --------------------------------------------------------------------------------------------
+
+	private static void testPutBytes(final byte[] heapMemory, final ByteBuffer offHeapMemory,
+										final int numValues, final int rounds) {
+		
+		TestRunner pureHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHeapMemorySegment seg = new PureHeapMemorySegment(heapMemory);
+				return timePutBytesOnHeap(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(heapMemory);
+				return timePutBytesHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(offHeapMemory);
+				return timePutBytesHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HeapMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timePutBytesAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timePutBytesAbstract(seg, numValues, rounds);
+			}
+		};
+		
+		TestRunner coreHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(offHeapMemory, null);
+				return timePutBytesAbstract(seg, numValues, rounds);
+			}
+		};
+		
+		TestRunner[] tests = {
+				TEST_CORE_ON_HEAP ? coreHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridOffHeapRunner : null,
+				pureHeapRunner, pureHybridHeapRunner, pureHybridOffHeapRunner
+		};
+
+		long[] results = runTestsInRandomOrder(tests, new Random(), 5, true);
+
+		System.out.println(String.format(
+				"Writing %d x %d bytes to %d bytes segment: " +
+						"\n\theap=%,d msecs" +
+						"\n\thybrid-on-heap=%,d msecs" +
+						"\n\thybrid-off-heap=%,d msecs" +
+						"\n\tspecialized heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-off-heap=%,d msecs, ",
+				rounds, numValues, heapMemory.length,
+				(results[0] / 1000000), (results[1] / 1000000), (results[2] / 1000000),
+				(results[3] / 1000000), (results[4] / 1000000), (results[5] / 1000000)));
+	}
+
+	private static void testGetBytes(final byte[] heapMemory, final ByteBuffer offHeapMemory,
+										final int numValues, final int rounds) {
+
+		TestRunner pureHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHeapMemorySegment seg = new PureHeapMemorySegment(heapMemory);
+				return timeGetBytesOnHeap(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(heapMemory);
+				return timeGetBytesHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(offHeapMemory);
+				return timeGetBytesHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HeapMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timeGetBytesAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timeGetBytesAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(offHeapMemory, null);
+				return timeGetBytesAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner[] tests = {
+				TEST_CORE_ON_HEAP ? coreHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridOffHeapRunner : null,
+				pureHeapRunner, pureHybridHeapRunner, pureHybridOffHeapRunner
+		};
+
+		long[] results = runTestsInRandomOrder(tests, new Random(), 5, true);
+
+		System.out.println(String.format(
+				"Reading %d x %d bytes from %d bytes segment: " +
+						"\n\theap=%,d msecs" +
+						"\n\thybrid-on-heap=%,d msecs" +
+						"\n\thybrid-off-heap=%,d msecs" +
+						"\n\tspecialized heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-off-heap=%,d msecs, ",
+				rounds, numValues, heapMemory.length,
+				(results[0] / 1000000), (results[1] / 1000000), (results[2] / 1000000),
+				(results[3] / 1000000), (results[4] / 1000000), (results[5] / 1000000)));
+	}
+
+	private static long timePutBytesOnHeap(final PureHeapMemorySegment segment, final int num, final int rounds) {
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.put(offset, (byte) i);
+				offset++;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+
+	private static long timePutBytesOffHeap(final PureOffHeapMemorySegment segment, final int num, final int rounds) {
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.put(offset, (byte) i);
+				offset++;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+
+	private static long timePutBytesHybrid(final PureHybridMemorySegment segment, final int num, final int rounds) {
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.put(offset, (byte) i);
+				offset++;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+
+	private static long timePutBytesAbstract(final MemorySegment segment, final int num, final int rounds) {
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.put(offset, (byte) i);
+				offset++;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timeGetBytesOnHeap(final PureHeapMemorySegment segment, final int num, final int rounds) {
+		long l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.get(offset);
+				offset++;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+
+	private static long timeGetBytesOffHeap(final PureOffHeapMemorySegment segment, final int num, final int rounds) {
+		long l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.get(offset);
+				offset++;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+
+	private static long timeGetBytesHybrid(final PureHybridMemorySegment segment, final int num, final int rounds) {
+		long l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.get(offset);
+				offset++;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+
+
+	private static long timeGetBytesAbstract(final MemorySegment segment, final int num, final int rounds) {
+		long l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.get(offset);
+				offset++;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+
+	// --------------------------------------------------------------------------------------------
+	//                                  LONGs
+	// --------------------------------------------------------------------------------------------
+	
+	private static void testPutLongs(final byte[] heapMemory, final ByteBuffer offHeapMemory,
+										final int numValues, final int rounds) {
+
+		TestRunner pureHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHeapMemorySegment seg = new PureHeapMemorySegment(heapMemory);
+				return timePutLongsOnHeap(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(heapMemory);
+				return timePutLongsHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(offHeapMemory);
+				return timePutLongsHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HeapMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timePutLongsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timePutLongsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(offHeapMemory, null);
+				return timePutLongsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner[] tests = {
+				TEST_CORE_ON_HEAP ? coreHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridOffHeapRunner : null,
+				pureHeapRunner, pureHybridHeapRunner, pureHybridOffHeapRunner
+		};
+
+		long[] results = runTestsInRandomOrder(tests, new Random(), 5, true);
+
+		System.out.println(String.format(
+				"Writing %d x %d longs to %d bytes segment: " +
+						"\n\theap=%,d msecs" +
+						"\n\thybrid-on-heap=%,d msecs" +
+						"\n\thybrid-off-heap=%,d msecs" +
+						"\n\tspecialized heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-off-heap=%,d msecs, ",
+				rounds, numValues, heapMemory.length,
+				(results[0] / 1000000), (results[1] / 1000000), (results[2] / 1000000),
+				(results[3] / 1000000), (results[4] / 1000000), (results[5] / 1000000)));
+	}
+	
+	private static void testGetLongs(final byte[] heapMemory, final ByteBuffer offHeapMemory,
+										final int numValues, final int rounds) {
+
+		TestRunner pureHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHeapMemorySegment seg = new PureHeapMemorySegment(heapMemory);
+				return timeGetLongsOnHeap(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(heapMemory);
+				return timeGetLongsHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(offHeapMemory);
+				return timeGetLongsHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HeapMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timeGetLongsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timeGetLongsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(offHeapMemory, null);
+				return timeGetLongsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner[] tests = {
+				TEST_CORE_ON_HEAP ? coreHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridOffHeapRunner : null,
+				pureHeapRunner, pureHybridHeapRunner, pureHybridOffHeapRunner
+		};
+
+		long[] results = runTestsInRandomOrder(tests, new Random(), 5, true);
+
+		System.out.println(String.format(
+				"Reading %d x %d longs from %d bytes segment: " +
+						"\n\theap=%,d msecs" +
+						"\n\thybrid-on-heap=%,d msecs" +
+						"\n\thybrid-off-heap=%,d msecs" +
+						"\n\tspecialized heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-off-heap=%,d msecs, ",
+				rounds, numValues, heapMemory.length,
+				(results[0] / 1000000), (results[1] / 1000000), (results[2] / 1000000),
+				(results[3] / 1000000), (results[4] / 1000000), (results[5] / 1000000)));
+	}
+	
+	private static long timePutLongsOnHeap(final PureHeapMemorySegment segment, final int num, final int rounds) {
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.putLong(offset, LONG_VALUE);
+				offset += 8;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timePutLongsOffHeap(final PureOffHeapMemorySegment segment, final int num, final int rounds) {
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.putLong(offset, LONG_VALUE);
+				offset += 8;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timePutLongsHybrid(final PureHybridMemorySegment segment, final int num, final int rounds) {
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.putLong(offset, LONG_VALUE);
+				offset += 8;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+
+	private static long timePutLongsAbstract(final MemorySegment segment, final int num, final int rounds) {
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.putLong(offset, LONG_VALUE);
+				offset += 8;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timeGetLongsOnHeap(final PureHeapMemorySegment segment, final int num, final int rounds) {
+		long l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.getLong(offset);
+				offset += 8;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+	
+	private static long timeGetLongsOffHeap(final PureOffHeapMemorySegment segment, final int num, final int rounds) {
+		long l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.getLong(offset);
+				offset += 8;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+	
+	private static long timeGetLongsHybrid(final PureHybridMemorySegment segment, final int num, final int rounds) {
+		// checked segment
+		long l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.getLong(offset);
+				offset += 8;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+
+	private static long timeGetLongsAbstract(final MemorySegment segment, final int num, final int rounds) {
+		// checked segment
+		long l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.getLong(offset);
+				offset += 8;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	//                                  INTs
+	// --------------------------------------------------------------------------------------------
+	
+	private static void testPutInts(final byte[] heapMemory, final ByteBuffer offHeapMemory,
+									final int numValues, final int rounds) {
+
+		TestRunner pureHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHeapMemorySegment seg = new PureHeapMemorySegment(heapMemory);
+				return timePutIntsOnHeap(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(heapMemory);
+				return timePutIntsHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(offHeapMemory);
+				return timePutIntsHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HeapMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timePutIntsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timePutIntsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(offHeapMemory, null);
+				return timePutIntsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner[] tests = {
+				TEST_CORE_ON_HEAP ? coreHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridOffHeapRunner : null,
+				pureHeapRunner, pureHybridHeapRunner, pureHybridOffHeapRunner
+		};
+
+		long[] results = runTestsInRandomOrder(tests, new Random(), 5, true);
+
+		System.out.println(String.format(
+				"Writing %d x %d ints to %d bytes segment: " +
+						"\n\theap=%,d msecs" +
+						"\n\thybrid-on-heap=%,d msecs" +
+						"\n\thybrid-off-heap=%,d msecs" +
+						"\n\tspecialized heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-off-heap=%,d msecs, ",
+				rounds, numValues, heapMemory.length,
+				(results[0] / 1000000), (results[1] / 1000000), (results[2] / 1000000),
+				(results[3] / 1000000), (results[4] / 1000000), (results[5] / 1000000)));
+	}
+	
+	private static void testGetInts(final byte[] heapMemory, final ByteBuffer offHeapMemory,
+									final int numValues, final int rounds) {
+
+		TestRunner pureHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHeapMemorySegment seg = new PureHeapMemorySegment(heapMemory);
+				return timeGetIntsOnHeap(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(heapMemory);
+				return timeGetIntsHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(offHeapMemory);
+				return timeGetIntsHybrid(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HeapMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timeGetIntsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timeGetIntsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(offHeapMemory, null);
+				return timeGetIntsAbstract(seg, numValues, rounds);
+			}
+		};
+
+		TestRunner[] tests = {
+				TEST_CORE_ON_HEAP ? coreHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridOffHeapRunner : null,
+				pureHeapRunner, pureHybridHeapRunner, pureHybridOffHeapRunner
+		};
+
+		long[] results = runTestsInRandomOrder(tests, new Random(), 5, true);
+
+		System.out.println(String.format(
+				"Reading %d x %d ints from %d bytes segment: " +
+						"\n\theap=%,d msecs" +
+						"\n\thybrid-on-heap=%,d msecs" +
+						"\n\thybrid-off-heap=%,d msecs" +
+						"\n\tspecialized heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-off-heap=%,d msecs, ",
+				rounds, numValues, heapMemory.length,
+				(results[0] / 1000000), (results[1] / 1000000), (results[2] / 1000000),
+				(results[3] / 1000000), (results[4] / 1000000), (results[5] / 1000000)));
+	}
+	
+	private static long timePutIntsOnHeap(final PureHeapMemorySegment segment, final int num, final int rounds) {
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.putInt(offset, i);
+				offset += 4;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timePutIntsOffHeap(final PureOffHeapMemorySegment segment, final int num, final int rounds) {
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.putInt(offset, i);
+				offset += 4;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timePutIntsHybrid(final PureHybridMemorySegment segment, final int num, final int rounds) {
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.putInt(offset, i);
+				offset += 4;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+
+	private static long timePutIntsAbstract(final MemorySegment segment, final int num, final int rounds) {
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.putInt(offset, i);
+				offset += 4;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timeGetIntsOnHeap(final PureHeapMemorySegment segment, final int num, final int rounds) {
+		int l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.getInt(offset);
+				offset += 4;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+	
+	private static long timeGetIntsOffHeap(final PureOffHeapMemorySegment segment, final int num, final int rounds) {
+		int l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.getInt(offset);
+				offset += 4;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+	
+	private static long timeGetIntsHybrid(final PureHybridMemorySegment segment, final int num, final int rounds) {
+		int l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.getInt(offset);
+				offset += 4;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+
+	private static long timeGetIntsAbstract(final MemorySegment segment, final int num, final int rounds) {
+		int l = 0;
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				l += segment.getInt(offset);
+				offset += 4;
+			}
+		}
+		long end = System.nanoTime();
+		sideEffect += l;
+		return end - start;
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	//                                  BYTE ARRAYs
+	// --------------------------------------------------------------------------------------------
+	
+	private static void testPutByteArrays1024(final byte[] heapMemory, final ByteBuffer offHeapMemory, 
+												final int numValues, final int rounds) {
+		
+		final byte[] sourceArray = new byte[1024];
+		for (int i = 0; i < sourceArray.length; i++) {
+			sourceArray[i] = (byte) i;
+		}
+
+		TestRunner pureHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHeapMemorySegment seg = new PureHeapMemorySegment(heapMemory);
+				return timePutByteArrayOnHeap(seg, sourceArray, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(heapMemory);
+				return timePutByteArrayHybrid(seg, sourceArray, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(offHeapMemory);
+				return timePutByteArrayHybrid(seg, sourceArray, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HeapMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timePutByteArrayAbstract(seg, sourceArray, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timePutByteArrayAbstract(seg, sourceArray, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(offHeapMemory, null);
+				return timePutByteArrayAbstract(seg, sourceArray, numValues, rounds);
+			}
+		};
+
+		TestRunner[] tests = {
+				TEST_CORE_ON_HEAP ? coreHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridOffHeapRunner : null,
+				pureHeapRunner, pureHybridHeapRunner, pureHybridOffHeapRunner
+		};
+
+		long[] results = runTestsInRandomOrder(tests, new Random(), 5, true);
+
+		System.out.println(String.format(
+				"Writing %d x %d byte[1024] to %d bytes segment: " +
+						"\n\theap=%,d msecs" +
+						"\n\thybrid-on-heap=%,d msecs" +
+						"\n\thybrid-off-heap=%,d msecs" +
+						"\n\tspecialized heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-off-heap=%,d msecs, ",
+				rounds, numValues, heapMemory.length,
+				(results[0] / 1000000), (results[1] / 1000000), (results[2] / 1000000),
+				(results[3] / 1000000), (results[4] / 1000000), (results[5] / 1000000)));
+	}
+	
+	private static void testGetByteArrays1024(final byte[] heapMemory, final ByteBuffer offHeapMemory,
+												final int numValues, final int rounds) {
+		
+		final byte[] targetArray = new byte[1024];
+
+		TestRunner pureHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHeapMemorySegment seg = new PureHeapMemorySegment(heapMemory);
+				return timeGetByteArrayOnHeap(seg, targetArray, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(heapMemory);
+				return timeGetByteArrayHybrid(seg, targetArray, numValues, rounds);
+			}
+		};
+
+		TestRunner pureHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				PureHybridMemorySegment seg = new PureHybridMemorySegment(offHeapMemory);
+				return timeGetByteArrayHybrid(seg, targetArray, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HeapMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timeGetByteArrayAbstract(seg, targetArray, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOnHeap(heapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledHeapMemory(heapMemory, null);
+				return timeGetByteArrayAbstract(seg, targetArray, numValues, rounds);
+			}
+		};
+
+		TestRunner coreHybridOffHeapRunner = new TestRunner() {
+			@Override
+			public long runTest() {
+				fillOffHeap(offHeapMemory, (byte) 0);
+				MemorySegment seg = HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(offHeapMemory, null);
+				return timeGetByteArrayAbstract(seg, targetArray, numValues, rounds);
+			}
+		};
+
+		TestRunner[] tests = {
+				TEST_CORE_ON_HEAP ? coreHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridHeapRunner : null,
+				TEST_CORE_OFF_HEAP ? coreHybridOffHeapRunner : null,
+				pureHeapRunner, pureHybridHeapRunner, pureHybridOffHeapRunner
+		};
+
+		long[] results = runTestsInRandomOrder(tests, new Random(), 5, true);
+
+		System.out.println(String.format(
+				"Reading %d x %d byte[1024] from %d bytes segment: " +
+						"\n\theap=%,d msecs" +
+						"\n\thybrid-on-heap=%,d msecs" +
+						"\n\thybrid-off-heap=%,d msecs" +
+						"\n\tspecialized heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-heap=%,d msecs, " +
+						"\n\tspecialized-hybrid-off-heap=%,d msecs, ",
+				rounds, numValues, heapMemory.length,
+				(results[0] / 1000000), (results[1] / 1000000), (results[2] / 1000000),
+				(results[3] / 1000000), (results[4] / 1000000), (results[5] / 1000000)));
+	}
+	
+	private static long timePutByteArrayOnHeap(final PureHeapMemorySegment segment, final byte[] source, final int num, final int rounds) {
+		final int len = source.length;
+		
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.put(offset, source, 0, len);
+				offset += len;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timePutByteArrayOffHeap(final PureOffHeapMemorySegment segment, final byte[] source, final int num, final int rounds) {
+		final int len = source.length;
+		
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.put(offset, source, 0, len);
+				offset += len;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timePutByteArrayHybrid(final PureHybridMemorySegment segment, final byte[] source, final int num, final int rounds) {
+		final int len = source.length;
+		
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.put(offset, source, 0, len);
+				offset += len;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+
+	private static long timePutByteArrayAbstract(final MemorySegment segment, final byte[] source, final int num, final int rounds) {
+		final int len = source.length;
+
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.put(offset, source, 0, len);
+				offset += len;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timeGetByteArrayOnHeap(final PureHeapMemorySegment segment, final byte[] target, final int num, final int rounds) {
+		final int len = target.length;
+		
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.get(offset, target, 0, len);
+				offset += len;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timeGetByteArrayOffHeap(final PureOffHeapMemorySegment segment, final byte[] target, final int num, final int rounds) {
+		final int len = target.length;
+		
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.get(offset, target, 0, len);
+				offset += len;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+	
+	private static long timeGetByteArrayHybrid(final PureHybridMemorySegment segment, final byte[] target, final int num, final int rounds) {
+		final int len = target.length;
+		
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.get(offset, target, 0, len);
+				offset += len;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+
+	private static long timeGetByteArrayAbstract(final MemorySegment segment, final byte[] target, final int num, final int rounds) {
+		final int len = target.length;
+
+		// checked segment
+		long start = System.nanoTime();
+		for (int round = 0; round < rounds; round++) {
+			int offset = 0;
+			for (int i = 0; i < num; i++) {
+				segment.get(offset, target, 0, len);
+				offset += len;
+			}
+		}
+		long end = System.nanoTime();
+		return end - start;
+	}
+
+//	// --------------------------------------------------------------------------------------------
+//	//                                  LONG BIG ENDIAN
+//	// --------------------------------------------------------------------------------------------
+//
+//	private static void testPutLongsBigEndian(byte[] heapMemory, ByteBuffer offHeapMemory, int numValues, int rounds) {
+//		// test the pure heap memory 
+//		fillOnHeap(heapMemory, (byte) 0);
+//		PureHeapMemorySegment heapMemorySegment = new PureHeapMemorySegment(heapMemory);
+//		long elapsedOnHeap = timePutLongsBigEndianOnHeap(heapMemorySegment, numValues, rounds);
+//
+//		// test the pure off-heap memory
+//		fillOffHeap(offHeapMemory, (byte) 0);
+//		PureOffHeapMemorySegment offHeapMemorySegment = new PureOffHeapMemorySegment(offHeapMemory);
+//		long elapsedOffHeap = timePutLongsBigEndianOffHeap(offHeapMemorySegment, numValues, rounds);
+//
+//		// test hybrid on heap 
+//		fillOnHeap(heapMemory, (byte) 0);
+//		PureHybridMemorySegment hybridOnHeap = new PureHybridMemorySegment(heapMemory);
+//		long elapsedHybridOnHeap = timePutLongsBigEndianHybrid(hybridOnHeap, numValues, rounds);
+//
+//		// test hybrid off heap 
+//		fillOffHeap(offHeapMemory, (byte) 0);
+//		PureHybridMemorySegment hybridOffeap = new PureHybridMemorySegment(offHeapMemory);
+//		long elapsedHybridOffHeap = timePutLongsBigEndianHybrid(hybridOffeap, numValues, rounds);
+//
+//		System.out.println(String.format(
+//				"Writing %d x %d big-endian longs to %d bytes segment: " +
+//						"heap=%,d msecs, " +
+//						"off-heap=%,d msecs, " +
+//						"hybrid-on-heap=%,d msecs, " +
+//						"hybrid-off-heap=%,d msecs",
+//				rounds, numValues, heapMemory.length,
+//				(elapsedOnHeap / 1000000), (elapsedOffHeap / 1000000),
+//				(elapsedHybridOnHeap / 1000000), (elapsedHybridOffHeap / 1000000)));
+//	}
+//
+//	private static void testGetLongsBigEndian(byte[] heapMemory, ByteBuffer offHeapMemory, int numValues, int rounds) {
+//		// test the pure heap memory 
+//		fillOnHeap(heapMemory, (byte) 0);
+//		PureHeapMemorySegment heapMemorySegment = new PureHeapMemorySegment(heapMemory);
+//		long elapsedOnHeap = timeGetLongsBigEndianOnHeap(heapMemorySegment, numValues, rounds);
+//
+//		// test the pure off-heap memory
+//		fillOffHeap(offHeapMemory, (byte) 0);
+//		PureOffHeapMemorySegment offHeapMemorySegment = new PureOffHeapMemorySegment(offHeapMemory);
+//		long elapsedOffHeap = timeGetLongsBigEndianOffHeap(offHeapMemorySegment, numValues, rounds);
+//
+//		// test hybrid on heap 
+//		fillOnHeap(heapMemory, (byte) 0);
+//		PureHybridMemorySegment hybridOnHeap = new PureHybridMemorySegment(heapMemory);
+//		long elapsedHybridOnHeap = timeGetLongsBigEndianHybrid(hybridOnHeap, numValues, rounds);
+//
+//		// test hybrid off heap 
+//		fillOffHeap(offHeapMemory, (byte) 0);
+//		PureHybridMemorySegment hybridOffeap = new PureHybridMemorySegment(offHeapMemory);
+//		long elapsedHybridOffHeap = timeGetLongsBigEndianHybrid(hybridOffeap, numValues, rounds);
+//
+//		System.out.println(String.format(
+//				"Reading %d x %d big-endian longs from %d bytes segment: " +
+//						"heap=%,d msecs, " +
+//						"off-heap=%,d msecs, " +
+//						"hybrid-on-heap=%,d msecs, " +
+//						"hybrid-off-heap=%,d msecs",
+//				rounds, numValues, heapMemory.length,
+//				(elapsedOnHeap / 1000000), (elapsedOffHeap / 1000000),
+//				(elapsedHybridOnHeap / 1000000), (elapsedHybridOffHeap / 1000000)));
+//	}
+//
+//	private static long timePutLongsBigEndianOnHeap(final PureHeapMemorySegment segment, final int num, final int rounds) {
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				segment.putLongBigEndian(offset, LONG_VALUE);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		return end - start;
+//	}
+//
+//	private static long timePutLongsBigEndianOffHeap(final PureOffHeapMemorySegment segment, final int num, final int rounds) {
+//		// checked segment
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				segment.putLongBigEndian(offset, LONG_VALUE);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		return end - start;
+//	}
+//
+//	private static long timePutLongsBigEndianHybrid(final PureHybridMemorySegment segment, final int num, final int rounds) {
+//		// checked segment
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				segment.putLongBigEndian(offset, LONG_VALUE);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		return end - start;
+//	}
+//
+//	private static long timeGetLongsBigEndianOnHeap(final PureHeapMemorySegment segment, final int num, final int rounds) {
+//		long l = 0;
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				l += segment.getLongBigEndian(offset);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		sideEffect += l;
+//		return end - start;
+//	}
+//
+//	private static long timeGetLongsBigEndianOffHeap(final PureOffHeapMemorySegment segment, final int num, final int rounds) {
+//		long l = 0;
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				l += segment.getLongBigEndian(offset);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		sideEffect += l;
+//		return end - start;
+//	}
+//
+//	private static long timeGetLongsBigEndianHybrid(final PureHybridMemorySegment segment, final int num, final int rounds) {
+//		// checked segment
+//		long l = 0;
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				l += segment.getLongBigEndian(offset);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		sideEffect += l;
+//		return end - start;
+//	}
+//
+//	// --------------------------------------------------------------------------------------------
+//	//                                  LONG LITTLE ENDIAN
+//	// --------------------------------------------------------------------------------------------
+//
+//	private static void testPutLongsLittleEndian(byte[] heapMemory, ByteBuffer offHeapMemory, int numValues, int rounds) {
+//		// test the pure heap memory 
+//		fillOnHeap(heapMemory, (byte) 0);
+//		PureHeapMemorySegment heapMemorySegment = new PureHeapMemorySegment(heapMemory);
+//		long elapsedOnHeap = timePutLongsLittleEndianOnHeap(heapMemorySegment, numValues, rounds);
+//
+//		// test the pure off-heap memory
+//		fillOffHeap(offHeapMemory, (byte) 0);
+//		PureOffHeapMemorySegment offHeapMemorySegment = new PureOffHeapMemorySegment(offHeapMemory);
+//		long elapsedOffHeap = timePutLongsLittleEndianOffHeap(offHeapMemorySegment, numValues, rounds);
+//
+//		// test hybrid on heap 
+//		fillOnHeap(heapMemory, (byte) 0);
+//		PureHybridMemorySegment hybridOnHeap = new PureHybridMemorySegment(heapMemory);
+//		long elapsedHybridOnHeap = timePutLongsLittleEndianHybrid(hybridOnHeap, numValues, rounds);
+//
+//		// test hybrid off heap 
+//		fillOffHeap(offHeapMemory, (byte) 0);
+//		PureHybridMemorySegment hybridOffeap = new PureHybridMemorySegment(offHeapMemory);
+//		long elapsedHybridOffHeap = timePutLongsLittleEndianHybrid(hybridOffeap, numValues, rounds);
+//
+//		System.out.println(String.format(
+//				"Writing %d x %d little-endian longs to %d bytes segment: " +
+//						"heap=%,d msecs, " +
+//						"off-heap=%,d msecs, " +
+//						"hybrid-on-heap=%,d msecs, " +
+//						"hybrid-off-heap=%,d msecs",
+//				rounds, numValues, heapMemory.length,
+//				(elapsedOnHeap / 1000000), (elapsedOffHeap / 1000000),
+//				(elapsedHybridOnHeap / 1000000), (elapsedHybridOffHeap / 1000000)));
+//	}
+//
+//	private static void testGetLongsLittleEndian(byte[] heapMemory, ByteBuffer offHeapMemory, int numValues, int rounds) {
+//		// test the pure heap memory 
+//		fillOnHeap(heapMemory, (byte) 0);
+//		PureHeapMemorySegment heapMemorySegment = new PureHeapMemorySegment(heapMemory);
+//		long elapsedOnHeap = timeGetLongsLittleEndianOnHeap(heapMemorySegment, numValues, rounds);
+//
+//		// test the pure off-heap memory
+//		fillOffHeap(offHeapMemory, (byte) 0);
+//		PureOffHeapMemorySegment offHeapMemorySegment = new PureOffHeapMemorySegment(offHeapMemory);
+//		long elapsedOffHeap = timeGetLongsLittleEndianOffHeap(offHeapMemorySegment, numValues, rounds);
+//
+//		// test hybrid on heap 
+//		fillOnHeap(heapMemory, (byte) 0);
+//		PureHybridMemorySegment hybridOnHeap = new PureHybridMemorySegment(heapMemory);
+//		long elapsedHybridOnHeap = timeGetLongsLittleEndianHybrid(hybridOnHeap, numValues, rounds);
+//
+//		// test hybrid off heap 
+//		fillOffHeap(offHeapMemory, (byte) 0);
+//		PureHybridMemorySegment hybridOffeap = new PureHybridMemorySegment(offHeapMemory);
+//		long elapsedHybridOffHeap = timeGetLongsLittleEndianHybrid(hybridOffeap, numValues, rounds);
+//
+//		System.out.println(String.format(
+//				"Reading %d x %d little-endian longs from %d bytes segment: " +
+//						"heap=%,d msecs, " +
+//						"off-heap=%,d msecs, " +
+//						"hybrid-on-heap=%,d msecs, " +
+//						"hybrid-off-heap=%,d msecs",
+//				rounds, numValues, heapMemory.length,
+//				(elapsedOnHeap / 1000000), (elapsedOffHeap / 1000000),
+//				(elapsedHybridOnHeap / 1000000), (elapsedHybridOffHeap / 1000000)));
+//	}
+//
+//	private static long timePutLongsLittleEndianOnHeap(final PureHeapMemorySegment segment, final int num, final int rounds) {
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				segment.putLongLittleEndian(offset, LONG_VALUE);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		return end - start;
+//	}
+//
+//	private static long timePutLongsLittleEndianOffHeap(final PureOffHeapMemorySegment segment, final int num, final int rounds) {
+//		// checked segment
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				segment.putLongLittleEndian(offset, LONG_VALUE);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		return end - start;
+//	}
+//
+//	private static long timePutLongsLittleEndianHybrid(final PureHybridMemorySegment segment, final int num, final int rounds) {
+//		// checked segment
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				segment.putLongLittleEndian(offset, LONG_VALUE);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		return end - start;
+//	}
+//
+//	private static long timeGetLongsLittleEndianOnHeap(final PureHeapMemorySegment segment, final int num, final int rounds) {
+//		long l = 0;
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				l += segment.getLongLittleEndian(offset);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		sideEffect += l;
+//		return end - start;
+//	}
+//
+//	private static long timeGetLongsLittleEndianOffHeap(final PureOffHeapMemorySegment segment, final int num, final int rounds) {
+//		long l = 0;
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				l += segment.getLongLittleEndian(offset);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		sideEffect += l;
+//		return end - start;
+//	}
+//
+//	private static long timeGetLongsLittleEndianHybrid(final PureHybridMemorySegment segment, final int num, final int rounds) {
+//		// checked segment
+//		long l = 0;
+//		long start = System.nanoTime();
+//		for (int round = 0; round < rounds; round++) {
+//			int offset = 0;
+//			for (int i = 0; i < num; i++) {
+//				l += segment.getLongLittleEndian(offset);
+//				offset += 8;
+//			}
+//		}
+//		long end = System.nanoTime();
+//		sideEffect += l;
+//		return end - start;
+//	}
+	
+	// ------------------------------------------------------------------------
+	//  Utilities
+	// ------------------------------------------------------------------------
+
+	private static void fillOnHeap(byte[] buffer, byte data) {
+		for (int i = 0; i < buffer.length; i++) {
+			buffer[i] = data;
+		}
+	}
+	
+	private static void fillOffHeap(ByteBuffer buffer, byte data) {
+		final int len = buffer.capacity();
+		for (int i = 0; i < len; i++) {
+			buffer.put(i, data);
+		}
+	}
+	
+	private static long[] runTestsInRandomOrder(TestRunner[] runners, Random rnd, int numRuns, boolean printMeasures) {
+		if (numRuns < 3) {
+			throw new IllegalArgumentException("must do at least three runs");
+		}
+		
+		// we run all runners in random order, to account for the JIT effects that specialize methods
+		// The observation is that either earlier tests suffer from performance because the JIT needs to kick
+		// in first, or that later tests suffer from performance, because the HIT optimized for the other case already
+		
+		long[][] measures = new long[runners.length][];
+		for (int i = 0; i < measures.length; i++) {
+			measures[i] = new long[numRuns];
+		}
+		
+		for (int test = 0; test < numRuns; test++) {
+			System.out.println("Round " + (test+1) + '/' + numRuns);
+			
+			// pick an order for the tests
+			int[] order = new int[runners.length];
+			for (int i = 0; i < order.length; i++) {
+				order[i] = i;
+			}
+			for (int i = order.length; i > 1; i--) {
+				int pos1 = i-1;
+				int pos2 = rnd.nextInt(i);
+				int tmp = order[pos1];
+				order[pos1] = order[pos2];
+				order[pos2] = tmp;
+			}
+			
+			// run tests
+			for (int pos : order) {
+				TestRunner next = runners[pos];
+				measures[pos][test] = next != null ? next.runTest() : 0L;
+			}
+		}
+		
+		if (printMeasures) {
+			for (long[] series : measures) {
+				StringBuilder bld = new StringBuilder();
+				for (long measure : series) {
+					bld.append(String.format("%,d", (measure / 1000000))).append(" | ");
+				}
+				System.out.println(bld.toString());
+			}
+		}
+		
+		// aggregate the measures
+		long[] results = new long[runners.length];
+		
+		for (int i = 0; i < runners.length; i++) {
+			// cancel out the min and max
+			long max = Long.MIN_VALUE;
+			long min = Long.MAX_VALUE;
+			
+			for (long val : measures[i]) {
+				max = Math.max(max, val);
+				min = Math.min(min, val);
+			}
+			
+			long total = 0L;
+			for (long val : measures[i]) {
+				if (val != max && val != min) {
+					total += val;
+				}
+			}
+			
+			results[i] = total / (numRuns - 2);
+		}
+		
+		return results;
+	}
+	
+	
+	
+	private static interface TestRunner {
+		
+		long runTest();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/PureHeapMemorySegment.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/PureHeapMemorySegment.java b/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/PureHeapMemorySegment.java
new file mode 100644
index 0000000..e247eed
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/PureHeapMemorySegment.java
@@ -0,0 +1,466 @@
+/*
+ * 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.benchmarks;
+
+import org.apache.flink.core.memory.MemoryUtils;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+public final class PureHeapMemorySegment {
+
+	/** Constant that flags the byte order. Because this is a boolean constant,
+	 * the JIT compiler can use this well to aggressively eliminate the non-applicable code paths */
+	private static final boolean LITTLE_ENDIAN = (ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN);
+
+	/** The array in which the data is stored. */
+	private byte[] memory;
+
+	/** Wrapper for I/O requests. */
+	private ByteBuffer wrapper;
+
+	/** The size, stored extra, because we may clear the reference to the byte array */
+	private final int size;
+
+	// -------------------------------------------------------------------------
+	//                             Constructors
+	// -------------------------------------------------------------------------
+
+	/**
+	 * Creates a new memory segment that represents the data in the given byte array.
+	 *
+	 * @param memory The byte array that holds the data.
+	 */
+	public PureHeapMemorySegment(byte[] memory) {
+		this.memory = memory;
+		this.size = memory.length;
+	}
+
+	// -------------------------------------------------------------------------
+	//                      Direct Memory Segment Specifics
+	// -------------------------------------------------------------------------
+
+	/**
+	 * Gets the byte array that backs this memory segment.
+	 *
+	 * @return The byte array that backs this memory segment.
+	 */
+	public byte[] getArray() {
+		return this.memory;
+	}
+
+	// -------------------------------------------------------------------------
+	//                        MemorySegment Accessors
+	// -------------------------------------------------------------------------
+	
+	public final boolean isFreed() {
+		return this.memory == null;
+	}
+	
+	public final void free() {
+		this.wrapper = null;
+		this.memory = null;
+	}
+	
+	public final int size() {
+		return this.size;
+	}
+	
+	public final ByteBuffer wrap(int offset, int length) {
+		if (offset > this.memory.length || offset > this.memory.length - length) {
+			throw new IndexOutOfBoundsException();
+		}
+
+		if (this.wrapper == null) {
+			this.wrapper = ByteBuffer.wrap(this.memory, offset, length);
+		}
+		else {
+			this.wrapper.limit(offset + length);
+			this.wrapper.position(offset);
+		}
+
+		return this.wrapper;
+	}
+	
+	// ------------------------------------------------------------------------
+	//                    Random Access get() and put() methods
+	// ------------------------------------------------------------------------
+	
+	public final byte get(int index) {
+		return this.memory[index];
+	}
+	
+	public final void put(int index, byte b) {
+		this.memory[index] = b;
+	}
+	
+	public final void get(int index, byte[] dst) {
+		get(index, dst, 0, dst.length);
+	}
+	
+	public final void put(int index, byte[] src) {
+		put(index, src, 0, src.length);
+	}
+	
+	public final void get(int index, byte[] dst, int offset, int length) {
+		// system arraycopy does the boundary checks anyways, no need to check extra
+		System.arraycopy(this.memory, index, dst, offset, length);
+	}
+	
+	public final void put(int index, byte[] src, int offset, int length) {
+		// system arraycopy does the boundary checks anyways, no need to check extra
+		System.arraycopy(src, offset, this.memory, index, length);
+	}
+	
+	public final boolean getBoolean(int index) {
+		return this.memory[index] != 0;
+	}
+	
+	public final void putBoolean(int index, boolean value) {
+		this.memory[index] = (byte) (value ? 1 : 0);
+	}
+	
+	@SuppressWarnings("restriction")
+	public final char getChar(int index) {
+		if (index >= 0 && index <= this.memory.length - 2) {
+			return UNSAFE.getChar(this.memory, BASE_OFFSET + index);
+		} else {
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	public final char getCharLittleEndian(int index) {
+		if (LITTLE_ENDIAN) {
+			return getChar(index);
+		} else {
+			return Character.reverseBytes(getChar(index));
+		}
+	}
+
+	public final char getCharBigEndian(int index) {
+		if (LITTLE_ENDIAN) {
+			return Character.reverseBytes(getChar(index));
+		} else {
+			return getChar(index);
+		}
+	}
+
+	@SuppressWarnings("restriction")
+	public final void putChar(int index, char value) {
+		if (index >= 0 && index <= this.memory.length - 2) {
+			UNSAFE.putChar(this.memory, BASE_OFFSET + index, value);
+		} else {
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	public final void putCharLittleEndian(int index, char value) {
+		if (LITTLE_ENDIAN) {
+			putChar(index, value);
+		} else {
+			putChar(index, Character.reverseBytes(value));
+		}
+	}
+
+	public final void putCharBigEndian(int index, char value) {
+		if (LITTLE_ENDIAN) {
+			putChar(index, Character.reverseBytes(value));
+		} else {
+			putChar(index, value);
+		}
+	}
+
+	@SuppressWarnings("restriction")
+	public final short getShort(int index) {
+		if (index >= 0 && index <= this.memory.length - 2) {
+			return UNSAFE.getShort(this.memory, BASE_OFFSET + index);
+		} else {
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	public final short getShortLittleEndian(int index) {
+		if (LITTLE_ENDIAN) {
+			return getShort(index);
+		} else {
+			return Short.reverseBytes(getShort(index));
+		}
+	}
+	
+	public final short getShortBigEndian(int index) {
+		if (LITTLE_ENDIAN) {
+			return Short.reverseBytes(getShort(index));
+		} else {
+			return getShort(index);
+		}
+	}
+
+	@SuppressWarnings("restriction")
+	public final void putShort(int index, short value) {
+		if (index >= 0 && index <= this.memory.length - 2) {
+			UNSAFE.putShort(this.memory, BASE_OFFSET + index, value);
+		} else {
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	public final void putShortLittleEndian(int index, short value) {
+		if (LITTLE_ENDIAN) {
+			putShort(index, value);
+		} else {
+			putShort(index, Short.reverseBytes(value));
+		}
+	}
+	
+	public final void putShortBigEndian(int index, short value) {
+		if (LITTLE_ENDIAN) {
+			putShort(index, Short.reverseBytes(value));
+		} else {
+			putShort(index, value);
+		}
+	}
+
+	@SuppressWarnings("restriction")
+	public final int getInt(int index) {
+		if (index >= 0 && index <= this.memory.length - 4) {
+			return UNSAFE.getInt(this.memory, BASE_OFFSET + index);
+		} else {
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	public final int getIntLittleEndian(int index) {
+		if (LITTLE_ENDIAN) {
+			return getInt(index);
+		} else {
+			return Integer.reverseBytes(getInt(index));
+		}
+	}
+	
+	public final int getIntBigEndian(int index) {
+		if (LITTLE_ENDIAN) {
+			return Integer.reverseBytes(getInt(index));
+		} else {
+			return getInt(index);
+		}
+	}
+	
+	@SuppressWarnings("restriction")
+	public final void putInt(int index, int value) {
+		if (index >= 0 && index <= this.memory.length - 4) {
+			UNSAFE.putInt(this.memory, BASE_OFFSET + index, value);
+		} else {
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	public final void putIntLittleEndian(int index, int value) {
+		if (LITTLE_ENDIAN) {
+			putInt(index, value);
+		} else {
+			putInt(index, Integer.reverseBytes(value));
+		}
+	}
+	
+	public final void putIntBigEndian(int index, int value) {
+		if (LITTLE_ENDIAN) {
+			putInt(index, Integer.reverseBytes(value));
+		} else {
+			putInt(index, value);
+		}
+	}
+
+	@SuppressWarnings("restriction")
+	public final long getLong(int index) {
+		if (index >= 0 && index <= this.memory.length - 8) {
+			return UNSAFE.getLong(this.memory, BASE_OFFSET + index);
+		} else {
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	public final long getLongLittleEndian(int index) {
+		if (LITTLE_ENDIAN) {
+			return getLong(index);
+		} else {
+			return Long.reverseBytes(getLong(index));
+		}
+	}
+	
+	public final long getLongBigEndian(int index) {
+		if (LITTLE_ENDIAN) {
+			return Long.reverseBytes(getLong(index));
+		} else {
+			return getLong(index);
+		}
+	}
+
+	@SuppressWarnings("restriction")
+	public final void putLong(int index, long value) {
+		if (index >= 0 && index <= this.memory.length - 8) {
+			UNSAFE.putLong(this.memory, BASE_OFFSET + index, value);
+		} else {
+			throw new IndexOutOfBoundsException();
+		}
+	}
+
+	public final void putLongLittleEndian(int index, long value) {
+		if (LITTLE_ENDIAN) {
+			putLong(index, value);
+		} else {
+			putLong(index, Long.reverseBytes(value));
+		}
+	}
+	
+	public final void putLongBigEndian(int index, long value) {
+		if (LITTLE_ENDIAN) {
+			putLong(index, Long.reverseBytes(value));
+		} else {
+			putLong(index, value);
+		}
+	}
+
+	public final float getFloat(int index) {
+		return Float.intBitsToFloat(getInt(index));
+	}
+	
+	public final float getFloatLittleEndian(int index) {
+		return Float.intBitsToFloat(getIntLittleEndian(index));
+	}
+	
+	public final float getFloatBigEndian(int index) {
+		return Float.intBitsToFloat(getIntBigEndian(index));
+	}
+	
+	public final void putFloat(int index, float value) {
+		putInt(index, Float.floatToRawIntBits(value));
+	}
+	
+	public final void putFloatLittleEndian(int index, float value) {
+		putIntLittleEndian(index, Float.floatToRawIntBits(value));
+	}
+	
+	public final void putFloatBigEndian(int index, float value) {
+		putIntBigEndian(index, Float.floatToRawIntBits(value));
+	}
+	
+	public final double getDouble(int index) {
+		return Double.longBitsToDouble(getLong(index));
+	}
+	
+	public final double getDoubleLittleEndian(int index) {
+		return Double.longBitsToDouble(getLongLittleEndian(index));
+	}
+
+	public final double getDoubleBigEndian(int index) {
+		return Double.longBitsToDouble(getLongBigEndian(index));
+	}
+	
+	public final void putDouble(int index, double value) {
+		putLong(index, Double.doubleToRawLongBits(value));
+	}
+	
+	public final void putDoubleLittleEndian(int index, double value) {
+		putLongLittleEndian(index, Double.doubleToRawLongBits(value));
+	}
+	
+	public final void putDoubleBigEndian(int index, double value) {
+		putLongBigEndian(index, Double.doubleToRawLongBits(value));
+	}
+
+	// -------------------------------------------------------------------------
+	//                     Bulk Read and Write Methods
+	// -------------------------------------------------------------------------
+	
+	public final void get(DataOutput out, int offset, int length) throws IOException {
+		out.write(this.memory, offset, length);
+	}
+	
+	public final void put(DataInput in, int offset, int length) throws IOException {
+		in.readFully(this.memory, offset, length);
+	}
+	
+	public final void get(int offset, ByteBuffer target, int numBytes) {
+		// ByteBuffer performs the boundary checks
+		target.put(this.memory, offset, numBytes);
+	}
+	
+	public final void put(int offset, ByteBuffer source, int numBytes) {
+		// ByteBuffer performs the boundary checks
+		source.get(this.memory, offset, numBytes);
+	}
+	
+	public final void copyTo(int offset, PureHeapMemorySegment target, int targetOffset, int numBytes) {
+		// system arraycopy does the boundary checks anyways, no need to check extra
+		System.arraycopy(this.memory, offset, target.memory, targetOffset, numBytes);
+	}
+
+	// -------------------------------------------------------------------------
+	//                      Comparisons & Swapping
+	// -------------------------------------------------------------------------
+	
+	public final int compare(PureHeapMemorySegment seg2, int offset1, int offset2, int len) {
+		final byte[] b2 = seg2.memory;
+		final byte[] b1 = this.memory;
+
+		int val = 0;
+		for (int pos = 0; pos < len && (val = (b1[offset1 + pos] & 0xff) - (b2[offset2 + pos] & 0xff)) == 0; pos++);
+		return val;
+	}
+
+	public final void swapBytes(PureHeapMemorySegment seg2, int offset1, int offset2, int len) {
+		// swap by bytes (chunks of 8 first, then single bytes)
+		while (len >= 8) {
+			long tmp = this.getLong(offset1);
+			this.putLong(offset1, seg2.getLong(offset2));
+			seg2.putLong(offset2, tmp);
+			offset1 += 8;
+			offset2 += 8;
+			len -= 8;
+		}
+		while (len > 0) {
+			byte tmp = this.get(offset1);
+			this.put(offset1, seg2.get(offset2));
+			seg2.put(offset2, tmp);
+			offset1++;
+			offset2++;
+			len--;
+		}
+	}
+	
+	public final void swapBytes(byte[] auxBuffer, PureHeapMemorySegment seg2, int offset1, int offset2, int len) {
+		byte[] otherMem = seg2.memory;
+		System.arraycopy(this.memory, offset1, auxBuffer, 0, len);
+		System.arraycopy(otherMem, offset2, this.memory, offset1, len);
+		System.arraycopy(auxBuffer, 0, otherMem, offset2, len);
+	}
+
+	// --------------------------------------------------------------------------------------------
+	//                     Utilities for native memory accesses and checks
+	// --------------------------------------------------------------------------------------------
+
+	@SuppressWarnings("restriction")
+	private static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
+
+	@SuppressWarnings("restriction")
+	private static final long BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/PureHeapMemorySegmentOutView.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/PureHeapMemorySegmentOutView.java b/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/PureHeapMemorySegmentOutView.java
new file mode 100644
index 0000000..1e3b89e
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/benchmarks/PureHeapMemorySegmentOutView.java
@@ -0,0 +1,359 @@
+/*
+ * 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.benchmarks;
+
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UTFDataFormatException;
+import java.util.List;
+
+public final class PureHeapMemorySegmentOutView implements DataOutputView {
+
+	private PureHeapMemorySegment currentSegment;	// the current memory segment to write to
+
+	private int positionInSegment;					// the offset in the current segment
+	
+	private final int segmentSize;				// the size of the memory segments
+
+	private final  List<PureHeapMemorySegment> memorySource;
+	
+	private final List<PureHeapMemorySegment> fullSegments;
+	
+
+	private byte[] utfBuffer;		// the reusable array for UTF encodings
+
+
+	public PureHeapMemorySegmentOutView(List<PureHeapMemorySegment> emptySegments,
+										List<PureHeapMemorySegment> fullSegmentTarget, int segmentSize) {
+		this.segmentSize = segmentSize;
+		this.currentSegment = emptySegments.remove(emptySegments.size() - 1);
+
+		this.memorySource = emptySegments;
+		this.fullSegments = fullSegmentTarget;
+		this.fullSegments.add(getCurrentSegment());
+	}
+
+
+	public void reset() {
+		if (this.fullSegments.size() != 0) {
+			throw new IllegalStateException("The target list still contains memory segments.");
+		}
+
+		clear();
+		try {
+			advance();
+		}
+		catch (IOException ioex) {
+			throw new RuntimeException("Error getting first segment for record collector.", ioex);
+		}
+	}
+	
+	// --------------------------------------------------------------------------------------------
+	//                                  Page Management
+	// --------------------------------------------------------------------------------------------
+
+	public PureHeapMemorySegment nextSegment(PureHeapMemorySegment current, int positionInCurrent) throws EOFException {
+		int size = this.memorySource.size();
+		if (size > 0) {
+			final PureHeapMemorySegment next = this.memorySource.remove(size - 1);
+			this.fullSegments.add(next);
+			return next;
+		} else {
+			throw new EOFException();
+		}
+	}
+	
+	public PureHeapMemorySegment getCurrentSegment() {
+		return this.currentSegment;
+	}
+
+	public int getCurrentPositionInSegment() {
+		return this.positionInSegment;
+	}
+	
+	public int getSegmentSize() {
+		return this.segmentSize;
+	}
+	
+	protected void advance() throws IOException {
+		this.currentSegment = nextSegment(this.currentSegment, this.positionInSegment);
+		this.positionInSegment = 0;
+	}
+	
+	protected void seekOutput(PureHeapMemorySegment seg, int position) {
+		this.currentSegment = seg;
+		this.positionInSegment = position;
+	}
+
+	protected void clear() {
+		this.currentSegment = null;
+		this.positionInSegment = 0;
+	}
+
+	// --------------------------------------------------------------------------------------------
+	//                               Data Output Specific methods
+	// --------------------------------------------------------------------------------------------
+
+	@Override
+	public void write(int b) throws IOException {
+		writeByte(b);
+	}
+
+	@Override
+	public void write(byte[] b) throws IOException {
+		write(b, 0, b.length);
+	}
+
+	@Override
+	public void write(byte[] b, int off, int len) throws IOException {
+		int remaining = this.segmentSize - this.positionInSegment;
+		if (remaining >= len) {
+			this.currentSegment.put(this.positionInSegment, b, off, len);
+			this.positionInSegment += len;
+		}
+		else {
+			if (remaining == 0) {
+				advance();
+				remaining = this.segmentSize - this.positionInSegment;
+			}
+			while (true) {
+				int toPut = Math.min(remaining, len);
+				this.currentSegment.put(this.positionInSegment, b, off, toPut);
+				off += toPut;
+				len -= toPut;
+
+				if (len > 0) {
+					this.positionInSegment = this.segmentSize;
+					advance();
+					remaining = this.segmentSize - this.positionInSegment;
+				}
+				else {
+					this.positionInSegment += toPut;
+					break;
+				}
+			}
+		}
+	}
+
+	@Override
+	public void writeBoolean(boolean v) throws IOException {
+		writeByte(v ? 1 : 0);
+	}
+
+	@Override
+	public void writeByte(int v) throws IOException {
+		if (this.positionInSegment < this.segmentSize) {
+			this.currentSegment.put(this.positionInSegment++, (byte) v);
+		}
+		else {
+			advance();
+			writeByte(v);
+		}
+	}
+
+	@Override
+	public void writeShort(int v) throws IOException {
+		if (this.positionInSegment < this.segmentSize - 1) {
+			this.currentSegment.putShortBigEndian(this.positionInSegment, (short) v);
+			this.positionInSegment += 2;
+		}
+		else if (this.positionInSegment == this.segmentSize) {
+			advance();
+			writeShort(v);
+		}
+		else {
+			writeByte(v >> 8);
+			writeByte(v);
+		}
+	}
+
+	@Override
+	public void writeChar(int v) throws IOException {
+		if (this.positionInSegment < this.segmentSize - 1) {
+			this.currentSegment.putCharBigEndian(this.positionInSegment, (char) v);
+			this.positionInSegment += 2;
+		}
+		else if (this.positionInSegment == this.segmentSize) {
+			advance();
+			writeChar(v);
+		}
+		else {
+			writeByte(v >> 8);
+			writeByte(v);
+		}
+	}
+
+	@Override
+	public void writeInt(int v) throws IOException {
+		if (this.positionInSegment < this.segmentSize - 3) {
+			this.currentSegment.putIntBigEndian(this.positionInSegment, v);
+			this.positionInSegment += 4;
+		}
+		else if (this.positionInSegment == this.segmentSize) {
+			advance();
+			writeInt(v);
+		}
+		else {
+			writeByte(v >> 24);
+			writeByte(v >> 16);
+			writeByte(v >>  8);
+			writeByte(v);
+		}
+	}
+
+	@Override
+	public void writeLong(long v) throws IOException {
+		if (this.positionInSegment < this.segmentSize - 7) {
+			this.currentSegment.putLongBigEndian(this.positionInSegment, v);
+			this.positionInSegment += 8;
+		}
+		else if (this.positionInSegment == this.segmentSize) {
+			advance();
+			writeLong(v);
+		}
+		else {
+			writeByte((int) (v >> 56));
+			writeByte((int) (v >> 48));
+			writeByte((int) (v >> 40));
+			writeByte((int) (v >> 32));
+			writeByte((int) (v >> 24));
+			writeByte((int) (v >> 16));
+			writeByte((int) (v >>  8));
+			writeByte((int) v);
+		}
+	}
+
+	@Override
+	public void writeFloat(float v) throws IOException {
+		writeInt(Float.floatToRawIntBits(v));
+	}
+
+	@Override
+	public void writeDouble(double v) throws IOException {
+		writeLong(Double.doubleToRawLongBits(v));
+	}
+
+	@Override
+	public void writeBytes(String s) throws IOException {
+		for (int i = 0; i < s.length(); i++) {
+			writeByte(s.charAt(i));
+		}
+	}
+
+	@Override
+	public void writeChars(String s) throws IOException {
+		for (int i = 0; i < s.length(); i++) {
+			writeChar(s.charAt(i));
+		}
+	}
+
+	@Override
+	public void writeUTF(String str) throws IOException {
+		int strlen = str.length();
+		int utflen = 0;
+		int c, count = 0;
+
+		/* use charAt instead of copying String to char array */
+		for (int i = 0; i < strlen; i++) {
+			c = str.charAt(i);
+			if ((c >= 0x0001) && (c <= 0x007F)) {
+				utflen++;
+			} else if (c > 0x07FF) {
+				utflen += 3;
+			} else {
+				utflen += 2;
+			}
+		}
+
+		if (utflen > 65535) {
+			throw new UTFDataFormatException("encoded string too long: " + utflen + " memory");
+		}
+
+		if (this.utfBuffer == null || this.utfBuffer.length < utflen + 2) {
+			this.utfBuffer = new byte[utflen + 2];
+		}
+		final byte[] bytearr = this.utfBuffer;
+
+		bytearr[count++] = (byte) ((utflen >>> 8) & 0xFF);
+		bytearr[count++] = (byte) (utflen & 0xFF);
+
+		int i = 0;
+		for (i = 0; i < strlen; i++) {
+			c = str.charAt(i);
+			if (!((c >= 0x0001) && (c <= 0x007F))) {
+				break;
+			}
+			bytearr[count++] = (byte) c;
+		}
+
+		for (; i < strlen; i++) {
+			c = str.charAt(i);
+			if ((c >= 0x0001) && (c <= 0x007F)) {
+				bytearr[count++] = (byte) c;
+
+			} else if (c > 0x07FF) {
+				bytearr[count++] = (byte) (0xE0 | ((c >> 12) & 0x0F));
+				bytearr[count++] = (byte) (0x80 | ((c >> 6) & 0x3F));
+				bytearr[count++] = (byte) (0x80 | (c & 0x3F));
+			} else {
+				bytearr[count++] = (byte) (0xC0 | ((c >> 6) & 0x1F));
+				bytearr[count++] = (byte) (0x80 | (c & 0x3F));
+			}
+		}
+
+		write(bytearr, 0, utflen + 2);
+	}
+
+	@Override
+	public void skipBytesToWrite(int numBytes) throws IOException {
+		while (numBytes > 0) {
+			final int remaining = this.segmentSize - this.positionInSegment;
+			if (numBytes <= remaining) {
+				this.positionInSegment += numBytes;
+				return;
+			}
+			this.positionInSegment = this.segmentSize;
+			advance();
+			numBytes -= remaining;
+		}
+	}
+
+	@Override
+	public void write(DataInputView source, int numBytes) throws IOException {
+		while (numBytes > 0) {
+			final int remaining = this.segmentSize - this.positionInSegment;
+			if (numBytes <= remaining) {
+				this.currentSegment.put(source, this.positionInSegment, numBytes);
+				this.positionInSegment += numBytes;
+				return;
+			}
+
+			if (remaining > 0) {
+				this.currentSegment.put(source, this.positionInSegment, remaining);
+				this.positionInSegment = this.segmentSize;
+				numBytes -= remaining;
+			}
+
+			advance();
+		}
+	}
+}


Mime
View raw message