flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-1320) Add an off-heap variant of the managed memory
Date Thu, 08 Jan 2015 11:41:34 GMT

    [ https://issues.apache.org/jira/browse/FLINK-1320?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14269233#comment-14269233
] 

ASF GitHub Bot commented on FLINK-1320:
---------------------------------------

Github user uce commented on a diff in the pull request:

    https://github.com/apache/flink/pull/290#discussion_r22647181
  
    --- 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");
    --- End diff --
    
    Did you forget these? For example the `AdaptiveSpanningRecordDeserializer` uses this method
and therefore wouldn't work with direct memory segments.


> 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)

Mime
View raw message