[
https://issues.apache.org/jira/browse/FLINK-1320?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14269341#comment-14269341
]
ASF GitHub Bot commented on FLINK-1320:
---------------------------------------
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/290#discussion_r22653183
--- Diff:
flink-core/src/main/java/org/apache/flink/core/memory/DirectMemorySegment.java
---
@@ -0,0 +1,560 @@
+/*
+ * 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 java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.BufferOverflowException;
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * This class uses in parts code from Java's direct byte buffer API.
+ *
+ * The use in this class two crucial additions:
+ * - It uses collapsed checks for range check and memory segment disposal.
+ * - It offers absolute positioning methods for byte array put/get
methods, to guarantee thread safe use.
+ *
+ * In addition, the code that uses this class should make sure that only
one implementation class is ever loaded -
+ * Either the {@link HeapMemorySegment}, or this DirectMemorySegment. That
way, all the abstract methods in the
+ * MemorySegment base class have only one loaded actual implementation.
This is easy for the JIT to recognize through
+ * class hierarchy analysis, or by identifying that the invocations are
monomorphic (all go to the same concrete
+ * method implementation). Under this precondition, the JIT can perfectly
inline methods.
+ *
+ * This is harder to do and control with byte buffers, where different
code paths use different versions of the class
+ * (heap, direct, mapped) and thus virtual method invocations are
polymorphic and are not as easily inlined.
+ */
+public class DirectMemorySegment extends MemorySegment {
+
+ /** The direct byte buffer that allocated the memory */
+ protected final ByteBuffer buffer;
+
+ /** The address to the off-heap data */
+ private long address;
+
+ /** The address one byte after the last addressable byte.
+ * This is address + size while the segment is not disposed */
+ private final long addressLimit;
+
+ /** The size in bytes of the memory segment */
+ private final int size;
+
+ //
-------------------------------------------------------------------------
+ // Constructors
+ //
-------------------------------------------------------------------------
+
+ public DirectMemorySegment(int size) {
+ this(ByteBuffer.allocateDirect(size));
+ }
+
+ public DirectMemorySegment(ByteBuffer buffer) {
+ if (buffer == null || !buffer.isDirect()) {
+ throw new IllegalArgumentException();
+ }
+
+ this.buffer = buffer;
+ this.size = buffer.capacity();
+ this.address = getAddress(buffer);
+ this.addressLimit = this.address + size;
+
+ if (address >= Long.MAX_VALUE - Integer.MAX_VALUE) {
+ throw new RuntimeException("Segment initialized with
too large address: " + address);
+ }
+ }
+
+ //
-------------------------------------------------------------------------
+ // MemorySegment Accessors
+ //
-------------------------------------------------------------------------
+
+
+ @Override
+ public final boolean isFreed() {
+ return this.address > this.addressLimit;
+ }
+
+ public final void free() {
+ // this ensures we can place no more data and trigger
+ // the checks for the freed segment
+ this.address = this.addressLimit + 1;
+ }
+
+ @Override
+ public final int size() {
+ return this.size;
+ }
+
+ @Override
+ public ByteBuffer wrap(int offset, int length) {
+ if (offset < 0 || offset > this.size || offset > this.size -
length) {
+ throw new IndexOutOfBoundsException();
+ }
+
+ this.buffer.limit(offset + length);
+ this.buffer.position(offset);
+
+ return this.buffer;
+ }
+
+
+ //
------------------------------------------------------------------------
+ // Random Access get() and put() methods
+ //
------------------------------------------------------------------------
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final byte get(int index) {
+
+ final long pos = address + index;
+ if (index >= 0 && pos < addressLimit) {
+ return UNSAFE.getByte(pos);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final void put(int index, byte b) {
+
+ final long pos = address + index;
+ if (index >= 0 && pos < addressLimit) {
+ UNSAFE.putByte(pos, b);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ public final void get(int index, byte[] dst) {
+ get(index, dst, 0, dst.length);
+ }
+
+ @Override
+ public final void put(int index, byte[] src) {
+ put(index, src, 0, src.length);
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final void get(int index, byte[] dst, int offset, int length) {
+
+ // check the byte array offset and length
+ if ((offset | length | (offset + length) | (dst.length -
(offset + length))) < 0) {
+ throw new IndexOutOfBoundsException();
+ }
+
+ long pos = address + index;
+
+ if (index >= 0 && pos <= addressLimit - length) {
+ long arrayAddress = BYTE_ARRAY_BASE_OFFSET + offset;
+
+ // the copy must proceed in batches not too large,
because the JVM may
+ // poll for points that are safe for GC (moving the
array and changing its address)
+ while (length > 0) {
+ long toCopy = (length > COPY_PER_BATCH) ?
COPY_PER_BATCH : length;
+ UNSAFE.copyMemory(null, pos, dst, arrayAddress,
toCopy);
+ length -= toCopy;
+ pos += toCopy;
+ arrayAddress += toCopy;
+ }
+ }
+ else if (address <= 0) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ 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();
+ }
+
+ long pos = address + index;
+ if (index >= 0 && pos <= addressLimit - length) {
+
+ long arrayAddress = BYTE_ARRAY_BASE_OFFSET + offset;
+ while (length > 0) {
+ long toCopy = (length > COPY_PER_BATCH) ?
COPY_PER_BATCH : length;
+ UNSAFE.copyMemory(src, arrayAddress, null, pos,
toCopy);
+ length -= toCopy;
+ pos += toCopy;
+ arrayAddress += toCopy;
+ }
+ }
+ else if (address <= 0) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ public final boolean getBoolean(int index) {
+ return get(index) != 0;
+ }
+
+ @Override
+ public final void putBoolean(int index, boolean value) {
+ put(index, (byte) (value ? 1 : 0));
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final char getChar(int index) {
+ final long pos = address + index;
+ if (index >= 0 && pos <= addressLimit - 2) {
+ return UNSAFE.getChar(pos);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final void putChar(int index, char value) {
+ final long pos = address + index;
+ if (index >= 0 && pos <= addressLimit - 2) {
+ UNSAFE.putChar(pos, value);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final short getShort(int index) {
+ final long pos = address + index;
+ if (index >= 0 && pos <= addressLimit - 2) {
+ return UNSAFE.getShort(pos);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final void putShort(int index, short value) {
+ final long pos = address + index;
+ if (index >= 0 && pos <= addressLimit - 2) {
+ UNSAFE.putShort(pos, value);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final int getInt(int index) {
+ final long pos = address + index;
+ if (index >= 0 && pos <= addressLimit - 4) {
+ return UNSAFE.getInt(pos);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final void putInt(int index, int value) {
+ final long pos = address + index;
+ if (index >= 0 && pos <= addressLimit - 4) {
+ UNSAFE.putInt(pos, value);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final long getLong(int index) {
+ final long pos = address + index;
+ if (index >= 0 && pos <= addressLimit - 8) {
+ return UNSAFE.getLong(pos);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final void putLong(int index, long value) {
+ final long pos = address + index;
+ if (index >= 0 && pos <= addressLimit - 8) {
+ UNSAFE.putLong(pos, value);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ // index is in fact invalid
+ throw new IndexOutOfBoundsException();
+ }
+ }
+
+ //
-------------------------------------------------------------------------
+ // Bulk Read and Write Methods
+ //
-------------------------------------------------------------------------
+
+ @Override
+ public final void get(DataOutput out, int offset, int length) throws
IOException {
+ throw new UnsupportedOperationException("not implemented");
+ }
+
+ @Override
+ public final void put(DataInput in, int offset, int length) throws
IOException {
+ throw new UnsupportedOperationException("not implemented");
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final void get(int offset, ByteBuffer target, int numBytes) {
+
+ // check the byte array offset and length
+ if ((offset | numBytes | (offset + numBytes) | (size - (offset
+ numBytes))) < 0) {
+ throw new IndexOutOfBoundsException();
+ }
+
+ final int targetOffset = target.position();
+ final int remaining = target.remaining();
+
+ if (remaining < numBytes) {
+ throw new BufferOverflowException();
+ }
+
+ if (target.isDirect()) {
+ // copy to the target memory directly
+ final long targetPointer = getAddress(target) +
targetOffset;
+ final long sourcePointer = address + offset;
+
+ if (sourcePointer <= addressLimit - numBytes) {
+ UNSAFE.copyMemory(sourcePointer, targetPointer,
numBytes);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ throw new IndexOutOfBoundsException();
+ }
+ }
+ else if (target.hasArray()) {
+ // move directly into the byte array
+ get(offset, target.array(), targetOffset +
target.arrayOffset(), numBytes);
+
+ // this must be after the get() call to ensue that the
byte buffer is not
+ // modified in case the call fails
+ target.position(targetOffset + numBytes);
+ }
+ else {
+ // neither heap buffer nor direct buffer
+ while (target.hasRemaining()) {
+ target.put(get(offset++));
+ }
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final void put(int offset, ByteBuffer source, int numBytes) {
+
+ // check the byte array offset and length
+ if ((offset | numBytes | (offset + numBytes) | (size - (offset
+ numBytes))) < 0) {
+ throw new IndexOutOfBoundsException();
+ }
+
+ final int sourceOffset = source.position();
+ final int remaining = source.remaining();
+
+ if (remaining < numBytes) {
+ throw new BufferUnderflowException();
+ }
+
+ if (source.isDirect()) {
+ // copy to the target memory directly
+ final long sourcePointer = getAddress(source) +
sourceOffset;
+ final long targetPointer = address + offset;
+
+ if (sourcePointer <= addressLimit - numBytes) {
+ UNSAFE.copyMemory(sourcePointer, targetPointer,
numBytes);
+ }
+ else if (address > addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ throw new IndexOutOfBoundsException();
+ }
+ }
+ else if (source.hasArray()) {
+ // move directly into the byte array
+ put(offset, source.array(), sourceOffset +
source.arrayOffset(), numBytes);
+
+ // this must be after the get() call to ensue that the
byte buffer is not
+ // modified in case the call fails
+ source.position(sourceOffset + numBytes);
+ }
+ else {
+ // neither heap buffer nor direct buffer
+ while (source.hasRemaining()) {
+ put(offset++, source.get());
+ }
+ }
+ }
+
+ @Override
+ @SuppressWarnings("restriction")
+ public final void copyTo(int offset, MemorySegment target, int
targetOffset, int numBytes) {
+ if (target.getClass() == DirectMemorySegment.class) {
+ DirectMemorySegment directOther = (DirectMemorySegment)
target;
+
+ final long thisPointer = address + offset;
+ final long otherPointer = directOther.address +
targetOffset;
+
+ if (numBytes >= 0 && thisPointer <= addressLimit -
numBytes && otherPointer <= directOther.addressLimit - numBytes) {
+ UNSAFE.copyMemory(thisPointer, otherPointer,
numBytes);
+ }
+ else if (address > addressLimit || directOther.address
> directOther.addressLimit) {
+ throw new IllegalStateException("disposed");
+ }
+ else {
+ throw new IndexOutOfBoundsException();
+ }
+ }
+ else {
+ throw new IllegalArgumentException("Can only copy to
other direct memory segments");
+ }
+ }
+
+ //
-------------------------------------------------------------------------
+ // Comparisons & Swapping
+ //
-------------------------------------------------------------------------
+
+ @Override
+ public int compare(MemorySegment seg2, int offset1, int offset2, int
len) {
+ ByteBuffer b1 = this.buffer;
+ ByteBuffer b2 = ((DirectMemorySegment)seg2).buffer;
+ return b1.compareTo(b2);
+ }
+
+ @Override
+ public void swapBytes(MemorySegment seg2, int offset1, int offset2, int
len) {
+ 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--;
+ }
+
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // Utilities for native memory accesses and checks
+ //
--------------------------------------------------------------------------------------------
+
+ @SuppressWarnings("restriction")
+ private static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
+
+ @SuppressWarnings("restriction")
+ private static final long BYTE_ARRAY_BASE_OFFSET =
UNSAFE.arrayBaseOffset(byte[].class);
+
+ private static final long COPY_PER_BATCH = 1024 * 1024;
+
+ private static final Field ADDRESS_FIELD;
+
+ static {
+ try {
+ ADDRESS_FIELD =
java.nio.Buffer.class.getDeclaredField("address");
+ ADDRESS_FIELD.setAccessible(true);
+ }
+ catch (Throwable t) {
+ throw new RuntimeException("Cannot initialize
DirectMemorySegment - direct memory not supported by Flink");
--- End diff --
Yes, this should only occur if the JVM does not support direct memory
allocation.
> Add an off-heap variant of the managed memory
> ---------------------------------------------
>
> Key: FLINK-1320
> URL: https://issues.apache.org/jira/browse/FLINK-1320
> Project: Flink
> Issue Type: Improvement
> Components: Local Runtime
> Reporter: Stephan Ewen
> Priority: Minor
>
> For (nearly) all memory that Flink accumulates (in the form of sort buffers,
> hash tables, caching), we use a special way of representing data serialized
> across a set of memory pages. The big work lies in the way the algorithms are
> implemented to operate on pages, rather than on objects.
> The core class for the memory is the {{MemorySegment}}, which has all methods
> to set and get primitives values efficiently. It is a somewhat simpler (and
> faster) variant of a HeapByteBuffer.
> As such, it should be straightforward to create a version where the memory
> segment is not backed by a heap byte[], but by memory allocated outside the
> JVM, in a similar way as the NIO DirectByteBuffers, or the Netty direct
> buffers do it.
> This may have multiple advantages:
> - We reduce the size of the JVM heap (garbage collected) and the number and
> size of long living alive objects. For large JVM sizes, this may improve
> performance quite a bit. Utilmately, we would in many cases reduce JVM size
> to 1/3 to 1/2 and keep the remaining memory outside the JVM.
> - We save copies when we move memory pages to disk (spilling) or through
> the network (shuffling / broadcasting / forward piping)
> The changes required to implement this are
> - Add a {{UnmanagedMemorySegment}} that only stores the memory adress as a
> long, and the segment size. It is initialized from a DirectByteBuffer.
> - Allow the MemoryManager to allocate these MemorySegments, instead of the
> current ones.
> - Make sure that the startup script pick up the mode and configure the heap
> size and the max direct memory properly.
> Since the MemorySegment is probably the most performance critical class in
> Flink, we must take care that we do this right. The following are critical
> considerations:
> - If we want both solutions (heap and off-heap) to exist side-by-side
> (configurable), we must make the base MemorySegment abstract and implement
> two versions (heap and off-heap).
> - To get the best performance, we need to make sure that only one class
> gets loaded (or at least ever used), to ensure optimal JIT de-virtualization
> and inlining.
> - We should carefully measure the performance of both variants. From
> previous micro benchmarks, I remember that individual byte accesses in
> DirectByteBuffers (off-heap) were slightly slower than on-heap, any larger
> accesses were equally good or slightly better.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
