[08/15] flink git commit: [FLINK-1320] [core] Add an off-heap variant of the managed memory

[se...@apache.org]

http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/AbstractPagedInputView.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/AbstractPagedInputView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/AbstractPagedInputView.java
deleted file mode 100644
index 23081c9..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/AbstractPagedInputView.java
+++ /dev/null
@@ -1,566 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.memorymanager;
-
-import java.io.EOFException;
-import java.io.IOException;
-import java.io.UTFDataFormatException;
-
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.MemorySegment;
-
-
-/**
- * The base class for all input views that are backed by multiple memory pages. This base class contains all
- * decoding methods to read data from a page and detect page boundary crossing. The concrete sub classes must
- * implement the methods to provide the next memory page once the boundary is crossed.
- */
-public abstract class AbstractPagedInputView implements DataInputView {
-	
-	private MemorySegment currentSegment;
-	
-	protected final int headerLength;				// the number of bytes to skip at the beginning of each segment
-	
-	private int positionInSegment;					// the offset in the current segment
-	
-	private int limitInSegment;						// the limit in the current segment before switching to the next
-	
-	private byte[] utfByteBuffer;					// reusable byte buffer for utf-8 decoding
-	private char[] utfCharBuffer;					// reusable char buffer for utf-8 decoding
-	
-	
-	// --------------------------------------------------------------------------------------------
-	//                                    Constructors
-	// --------------------------------------------------------------------------------------------
-	
-	/**
-	 * Creates a new view that starts with the given segment. The input starts directly after the header
-	 * of the given page. If the header size is zero, it starts at the beginning. The specified initial
-	 * limit describes up to which position data may be read from the current segment, before the view must
-	 * advance to the next segment.
-	 * 
-	 * @param initialSegment The memory segment to start reading from.
-	 * @param initialLimit The position one after the last valid byte in the initial segment.
-	 * @param headerLength The number of bytes to skip at the beginning of each segment for the header. This
-	 *                     length must be the same for all memory segments.
-	 */
-	protected AbstractPagedInputView(MemorySegment initialSegment, int initialLimit, int headerLength) {
-		this.headerLength = headerLength;
-		this.positionInSegment = headerLength;
-		seekInput(initialSegment, headerLength, initialLimit);
-	}
-	
-	/**
-	 * Creates a new view that is initially not bound to a memory segment. This constructor is typically
-	 * for views that always seek first.
-	 * <p>
-	 * WARNING: The view is not readable until the first call to either {@link #advance()}, 
-	 * or to {@link #seekInput(MemorySegment, int, int)}.
-	 * 
-	 * @param headerLength The number of bytes to skip at the beginning of each segment for the header.
-	 */
-	protected AbstractPagedInputView(int headerLength) {
-		this.headerLength = headerLength;
-	}
-
-	// --------------------------------------------------------------------------------------------
-	//                                  Page Management
-	// --------------------------------------------------------------------------------------------
-	
-	/**
-	 * Gets the memory segment that will be used to read the next bytes from. If the segment is exactly exhausted,
-	 * meaning that the last byte read was the last byte available in the segment, then this segment will
-	 * not serve the next bytes. The segment to serve the next bytes will be obtained through the
-	 * {@link #nextSegment(MemorySegment)} method.
-	 * 
-	 * @return The current memory segment.
-	 */
-	public MemorySegment getCurrentSegment() {
-		return this.currentSegment;
-	}
-	
-	/**
-	 * Gets the position from which the next byte will be read. If that position is equal to the current limit,
-	 * then the next byte will be read from next segment.
-	 * 
-	 * @return The position from which the next byte will be read.
-	 * @see #getCurrentSegmentLimit()
-	 */
-	public int getCurrentPositionInSegment() {
-		return this.positionInSegment;
-	}
-	
-	/**
-	 * Gets the current limit in the memory segment. This value points to the byte one after the last valid byte
-	 * in the memory segment.
-	 * 
-	 * @return The current limit in the memory segment.
-	 * @see #getCurrentPositionInSegment()
-	 */
-	public int getCurrentSegmentLimit() {
-		return this.limitInSegment;
-	}
-	
-	/**
-	 * The method by which concrete subclasses realize page crossing. This method is invoked when the current page
-	 * is exhausted and a new page is required to continue the reading. If no further page is available, this
-	 * method must throw an {@link EOFException}.
-	 *  
-	 * @param current The current page that was read to its limit. May be {@code null}, if this method is
-	 *                invoked for the first time.
-	 * @return The next page from which the reading should continue. May not be {@code null}. If the input is
-	 *         exhausted, an {@link EOFException} must be thrown instead.
-	 *         
-	 * @throws EOFException Thrown, if no further segment is available.
-	 * @throws IOException Thrown, if the method cannot provide the next page due to an I/O related problem.
-	 */
-	protected abstract MemorySegment nextSegment(MemorySegment current) throws EOFException, IOException;
-	
-	/**
-	 * Gets the limit for reading bytes from the given memory segment. This method must return the position
-	 * of the byte after the last valid byte in the given memory segment. When the position returned by this
-	 * method is reached, the view will attempt to switch to the next memory segment.
-	 * 
-	 * @param segment The segment to determine the limit for.
-	 * @return The limit for the given memory segment.
-	 */
-	protected abstract int getLimitForSegment(MemorySegment segment);
-	
-	/**
-	 * Advances the view to the next memory segment. The reading will continue after the header of the next
-	 * segment. This method uses {@link #nextSegment(MemorySegment)} and {@link #getLimitForSegment(MemorySegment)}
-	 * to get the next segment and set its limit.
-	 * 
-	 * @throws IOException Thrown, if the next segment could not be obtained.
-	 * 
-	 * @see #nextSegment(MemorySegment)
-	 * @see #getLimitForSegment(MemorySegment)
-	 */
-	protected final void advance() throws IOException {
-		// note: this code ensures that in case of EOF, we stay at the same position such that
-		// EOF is reproducible (if nextSegment throws a reproducible EOFException)
-		this.currentSegment = nextSegment(this.currentSegment);
-		this.limitInSegment = getLimitForSegment(this.currentSegment);
-		this.positionInSegment = this.headerLength;
-	}
-	
-	/**
-	 * Sets the internal state of the view such that the next bytes will be read from the given memory segment,
-	 * starting at the given position. The memory segment will provide bytes up to the given limit position.
-	 * 
-	 * @param segment The segment to read the next bytes from.
-	 * @param positionInSegment The position in the segment to start reading from.
-	 * @param limitInSegment The limit in the segment. When reached, the view will attempt to switch to
-	 *                       the next segment.
-	 */
-	protected void seekInput(MemorySegment segment, int positionInSegment, int limitInSegment) {
-		this.currentSegment = segment;
-		this.positionInSegment = positionInSegment;
-		this.limitInSegment = limitInSegment;
-	}
-	
-	/**
-	 * Clears the internal state of the view. After this call, all read attempts will fail, until the
-	 * {@link #advance()} or {@link #seekInput(MemorySegment, int, int)} method have been invoked.
-	 */
-	protected void clear() {
-		this.currentSegment = null;
-		this.positionInSegment = this.headerLength;
-		this.limitInSegment = headerLength;
-	}
-	
-	// --------------------------------------------------------------------------------------------
-	//                               Data Input Specific methods
-	// --------------------------------------------------------------------------------------------
-
-	@Override
-	public int read(byte[] b) throws IOException{
-		return read(b,0,b.length);
-	}
-
-	@Override
-	public int read(byte[] b, int off, int len) throws IOException{
-		if (off < 0 || len < 0 || off + len > b.length) {
-			throw new IndexOutOfBoundsException();
-		}
-
-		int remaining = this.limitInSegment - this.positionInSegment;
-		if (remaining >= len) {
-			this.currentSegment.get(this.positionInSegment, b, off, len);
-			this.positionInSegment += len;
-			return len;
-		}
-		else {
-			if (remaining == 0) {
-				try {
-					advance();
-				}catch(EOFException eof){
-					return -1;
-				}
-				remaining = this.limitInSegment - this.positionInSegment;
-			}
-
-			int bytesRead = 0;
-			while (true) {
-				int toRead = Math.min(remaining, len-bytesRead);
-				this.currentSegment.get(this.positionInSegment, b, off, toRead);
-				off += toRead;
-				bytesRead += toRead;
-
-				if (len > bytesRead) {
-					try {
-						advance();
-					}catch(EOFException eof){
-						this.positionInSegment += toRead;
-						return bytesRead;
-					}
-					remaining = this.limitInSegment - this.positionInSegment;
-				}
-				else {
-					this.positionInSegment += toRead;
-					break;
-				}
-			}
-			return len;
-		}
-	}
-	
-	@Override
-	public void readFully(byte[] b) throws IOException {
-		readFully(b, 0, b.length);
-	}
-
-	@Override
-	public void readFully(byte[] b, int off, int len) throws IOException {
-		int bytesRead = read(b,off,len);
-
-		if(bytesRead < len){
-			throw new EOFException("There is no enough data left in the DataInputView.");
-		}
-	}
-
-	@Override
-	public boolean readBoolean() throws IOException {
-		return readByte() == 1;
-	}
-
-	@Override
-	public byte readByte() throws IOException {
-		if (this.positionInSegment < this.limitInSegment) {
-			return this.currentSegment.get(this.positionInSegment++);
-		}
-		else {
-			advance();
-			return readByte();
-		}
-	}
-
-	@Override
-	public int readUnsignedByte() throws IOException {
-		return readByte() & 0xff;
-	}
-
-	@Override
-	public short readShort() throws IOException {
-		if (this.positionInSegment < this.limitInSegment - 1) {
-			final short v = this.currentSegment.getShort(this.positionInSegment);
-			this.positionInSegment += 2;
-			return v;
-		}
-		else if (this.positionInSegment == this.limitInSegment) {
-			advance();
-			return readShort();
-		}
-		else {
-			return (short) ((readUnsignedByte() << 8) | readUnsignedByte());
-		}
-	}
-
-	@Override
-	public int readUnsignedShort() throws IOException {
-		if (this.positionInSegment < this.limitInSegment - 1) {
-			final int v = this.currentSegment.getShort(this.positionInSegment) & 0xffff;
-			this.positionInSegment += 2;
-			return v;
-		}
-		else if (this.positionInSegment == this.limitInSegment) {
-			advance();
-			return readUnsignedShort();
-		}
-		else {
-			return (readUnsignedByte() << 8) | readUnsignedByte();
-		}
-	}
-
-	@Override
-	public char readChar() throws IOException  {
-		if (this.positionInSegment < this.limitInSegment - 1) {
-			final char v = this.currentSegment.getChar(this.positionInSegment);
-			this.positionInSegment += 2;
-			return v;
-		}
-		else if (this.positionInSegment == this.limitInSegment) {
-			advance();
-			return readChar();
-		}
-		else {
-			return (char) ((readUnsignedByte() << 8) | readUnsignedByte());
-		}
-	}
-
-	@Override
-	public int readInt() throws IOException {
-		if (this.positionInSegment < this.limitInSegment - 3) {
-			final int v = this.currentSegment.getIntBigEndian(this.positionInSegment);
-			this.positionInSegment += 4;
-			return v;
-		}
-		else if (this.positionInSegment == this.limitInSegment) {
-			advance();
-			return readInt();
-		}
-		else {
-			return (readUnsignedByte() << 24) |
-				(readUnsignedByte() << 16) |
-				(readUnsignedByte() <<  8) |
-					readUnsignedByte();
-		}
-	}
-
-	@Override
-	public long readLong() throws IOException {
-		if (this.positionInSegment < this.limitInSegment - 7) {
-			final long v = this.currentSegment.getLongBigEndian(this.positionInSegment);
-			this.positionInSegment += 8;
-			return v;
-		}
-		else if (this.positionInSegment == this.limitInSegment) {
-			advance();
-			return readLong();
-		}
-		else {
-			long l = 0L;
-			l |= ((long) readUnsignedByte()) << 56;
-			l |= ((long) readUnsignedByte()) << 48;
-			l |= ((long) readUnsignedByte()) << 40;
-			l |= ((long) readUnsignedByte()) << 32;
-			l |= ((long) readUnsignedByte()) << 24;
-			l |= ((long) readUnsignedByte()) << 16;
-			l |= ((long) readUnsignedByte()) <<  8;
-			l |= (long) readUnsignedByte();
-			return l;
-		}
-	}
-
-	@Override
-	public float readFloat() throws IOException {
-		return Float.intBitsToFloat(readInt());
-	}
-
-	@Override
-	public double readDouble() throws IOException {
-		return Double.longBitsToDouble(readLong());
-	}
-
-	@Override
-	public String readLine() throws IOException {
-		final StringBuilder bld = new StringBuilder(32);
-		
-		try {
-			int b;
-			while ((b = readUnsignedByte()) != '\n') {
-				if (b != '\r') {
-					bld.append((char) b);
-				}
-			}
-		}
-		catch (EOFException eofex) {}
-
-		if (bld.length() == 0) {
-			return null;
-		}
-		
-		// trim a trailing carriage return
-		int len = bld.length();
-		if (len > 0 && bld.charAt(len - 1) == '\r') {
-			bld.setLength(len - 1);
-		}
-		return bld.toString();
-	}
-
-	@Override
-	public String readUTF() throws IOException {
-		final int utflen = readUnsignedShort();
-		
-		final byte[] bytearr;
-		final char[] chararr;
-		
-		if (this.utfByteBuffer == null || this.utfByteBuffer.length < utflen) {
-			bytearr = new byte[utflen];
-			this.utfByteBuffer = bytearr;
-		} else {
-			bytearr = this.utfByteBuffer;
-		}
-		if (this.utfCharBuffer == null || this.utfCharBuffer.length < utflen) {
-			chararr = new char[utflen];
-			this.utfCharBuffer = chararr;
-		} else {
-			chararr = this.utfCharBuffer;
-		}
-
-		int c, char2, char3;
-		int count = 0;
-		int chararr_count = 0;
-
-		readFully(bytearr, 0, utflen);
-
-		while (count < utflen) {
-			c = (int) bytearr[count] & 0xff;
-			if (c > 127) {
-				break;
-			}
-			count++;
-			chararr[chararr_count++] = (char) c;
-		}
-
-		while (count < utflen) {
-			c = (int) bytearr[count] & 0xff;
-			switch (c >> 4) {
-			case 0:
-			case 1:
-			case 2:
-			case 3:
-			case 4:
-			case 5:
-			case 6:
-			case 7:
-				/* 0xxxxxxx */
-				count++;
-				chararr[chararr_count++] = (char) c;
-				break;
-			case 12:
-			case 13:
-				/* 110x xxxx 10xx xxxx */
-				count += 2;
-				if (count > utflen) {
-					throw new UTFDataFormatException("malformed input: partial character at end");
-				}
-				char2 = (int) bytearr[count - 1];
-				if ((char2 & 0xC0) != 0x80) {
-					throw new UTFDataFormatException("malformed input around byte " + count);
-				}
-				chararr[chararr_count++] = (char) (((c & 0x1F) << 6) | (char2 & 0x3F));
-				break;
-			case 14:
-				/* 1110 xxxx 10xx xxxx 10xx xxxx */
-				count += 3;
-				if (count > utflen) {
-					throw new UTFDataFormatException("malformed input: partial character at end");
-				}
-				char2 = (int) bytearr[count - 2];
-				char3 = (int) bytearr[count - 1];
-				if (((char2 & 0xC0) != 0x80) || ((char3 & 0xC0) != 0x80)) {
-					throw new UTFDataFormatException("malformed input around byte " + (count - 1));
-				}
-				chararr[chararr_count++] = (char) (((c & 0x0F) << 12) | ((char2 & 0x3F) << 6) | ((char3 & 0x3F) << 0));
-				break;
-			default:
-				/* 10xx xxxx, 1111 xxxx */
-				throw new UTFDataFormatException("malformed input around byte " + count);
-			}
-		}
-		// The number of chars produced may be less than utflen
-		return new String(chararr, 0, chararr_count);
-	}
-	
-	@Override
-	public int skipBytes(int n) throws IOException {
-		if (n < 0) {
-			throw new IllegalArgumentException();
-		}
-		
-		int remaining = this.limitInSegment - this.positionInSegment;
-		if (remaining >= n) {
-			this.positionInSegment += n;
-			return n;
-		}
-		else {
-			if (remaining == 0) {
-				try {
-					advance();
-				} catch (EOFException eofex) {
-					return 0;
-				}
-				remaining = this.limitInSegment - this.positionInSegment;
-			}
-			
-			int skipped = 0;
-			while (true) {
-				int toSkip = Math.min(remaining, n);
-				n -= toSkip;
-				skipped += toSkip;
-				
-				if (n > 0) {
-					try {
-						advance();
-					} catch (EOFException eofex) {
-						return skipped;
-					}
-					remaining = this.limitInSegment - this.positionInSegment;	
-				}
-				else {
-					this.positionInSegment += toSkip;
-					break;
-				}
-			}
-			return skipped;
-		}
-	}
-
-	@Override
-	public void skipBytesToRead(int numBytes) throws IOException {
-		if (numBytes < 0) {
-			throw new IllegalArgumentException();
-		}
-		
-		int remaining = this.limitInSegment - this.positionInSegment;
-		if (remaining >= numBytes) {
-			this.positionInSegment += numBytes;
-		}
-		else {
-			if (remaining == 0) {
-				advance();
-				remaining = this.limitInSegment - this.positionInSegment;
-			}
-			
-			while (true) {
-				if (numBytes > remaining) {
-					numBytes -= remaining;
-					advance();
-					remaining = this.limitInSegment - this.positionInSegment;	
-				}
-				else {
-					this.positionInSegment += numBytes;
-					break;
-				}
-			}
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/AbstractPagedOutputView.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/AbstractPagedOutputView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/AbstractPagedOutputView.java
deleted file mode 100644
index ea73e62..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/AbstractPagedOutputView.java
+++ /dev/null
@@ -1,416 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.memorymanager;
-
-import java.io.IOException;
-import java.io.UTFDataFormatException;
-
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
-import org.apache.flink.core.memory.MemorySegment;
-
-
-/**
- * The base class for all output views that are backed by multiple memory pages. This base class contains all
- * encoding methods to write data to a page and detect page boundary crossing. The concrete sub classes must
- * implement the methods to collect the current page and provide the next memory page once the boundary is crossed.
- * <p>
- * The paging assumes that all memory segments are of the same size.
- */
-public abstract class AbstractPagedOutputView implements DataOutputView {
-	
-	private MemorySegment currentSegment;			// the current memory segment to write to
-	
-	protected final int segmentSize;				// the size of the memory segments
-	
-	protected final int headerLength;				// the number of bytes to skip at the beginning of each segment
-	
-	private int positionInSegment;					// the offset in the current segment
-	
-	private byte[] utfBuffer;						// the reusable array for UTF encodings
-	
-	
-	// --------------------------------------------------------------------------------------------
-	//                                    Constructors
-	// --------------------------------------------------------------------------------------------
-	
-	/**
-	 * Creates a new output view that writes initially to the given initial segment. All segments in the
-	 * view have to be of the given {@code segmentSize}. A header of length {@code headerLength} is left
-	 * at the beginning of each segment.
-	 * 
-	 * @param initialSegment The segment that the view starts writing to.
-	 * @param segmentSize The size of the memory segments.
-	 * @param headerLength The number of bytes to skip at the beginning of each segment for the header.
-	 */
-	protected AbstractPagedOutputView(MemorySegment initialSegment, int segmentSize, int headerLength) {
-		if (initialSegment == null) {
-			throw new NullPointerException("Initial Segment may not be null");
-		}
-		this.segmentSize = segmentSize;
-		this.headerLength = headerLength;
-		this.currentSegment = initialSegment;
-		this.positionInSegment = headerLength;
-	}
-	
-	/**
-	 * @param segmentSize The size of the memory segments.
-	 * @param headerLength The number of bytes to skip at the beginning of each segment for the header.
-	 */
-	protected AbstractPagedOutputView(int segmentSize, int headerLength)
-	{
-		this.segmentSize = segmentSize;
-		this.headerLength = headerLength;
-	}
-	
-
-	// --------------------------------------------------------------------------------------------
-	//                                  Page Management
-	// --------------------------------------------------------------------------------------------
-	
-	/**
-	 * 
-	 * This method must return a segment. If no more segments are available, it must throw an
-	 * {@link java.io.EOFException}.
-	 * 
-	 * @param current The current memory segment
-	 * @param positionInCurrent The position in the segment, one after the last valid byte.
-	 * @return The next memory segment. 
-	 * 
-	 * @throws IOException
-	 */
-	protected abstract MemorySegment nextSegment(MemorySegment current, int positionInCurrent) throws IOException;
-	
-	
-	/**
-	 * Gets the segment that the view currently writes to.
-	 * 
-	 * @return The segment the view currently writes to.
-	 */
-	public MemorySegment getCurrentSegment() {
-		return this.currentSegment;
-	}
-	
-	/**
-	 * Gets the current write position (the position where the next bytes will be written)
-	 * in the current memory segment.
-	 * 
-	 * @return The current write offset in the current memory segment.
-	 */
-	public int getCurrentPositionInSegment() {
-		return this.positionInSegment;
-	}
-	
-	/**
-	 * Gets the size of the segments used by this view.
-	 * 
-	 * @return The memory segment size.
-	 */
-	public int getSegmentSize() {
-		return this.segmentSize;
-	}
-	
-	/**
-	 * Moves the output view to the next page. This method invokes internally the
-	 * {@link #nextSegment(MemorySegment, int)} method to give the current memory segment to the concrete subclass' 
-	 * implementation and obtain the next segment to write to. Writing will continue inside the new segment
-	 * after the header.
-	 * 
-	 * @throws IOException Thrown, if the current segment could not be processed or a new segment could not
-	 *                     be obtained. 
-	 */
-	protected void advance() throws IOException {
-		this.currentSegment = nextSegment(this.currentSegment, this.positionInSegment);
-		this.positionInSegment = this.headerLength;
-	}
-	
-	/**
-	 * Sets the internal state to the given memory segment and the given position within the segment. 
-	 * 
-	 * @param seg The memory segment to write the next bytes to.
-	 * @param position The position to start writing the next bytes to.
-	 */
-	protected void seekOutput(MemorySegment seg, int position) {
-		this.currentSegment = seg;
-		this.positionInSegment = position;
-	}
-	
-	/**
-	 * Clears the internal state. Any successive write calls will fail until either {@link #advance()} or
-	 * {@link #seekOutput(MemorySegment, int)} is called. 
-	 * 
-	 * @see #advance()
-	 * @see #seekOutput(MemorySegment, int)
-	 */
-	protected void clear() {
-		this.currentSegment = null;
-		this.positionInSegment = this.headerLength;
-	}
-	
-	// --------------------------------------------------------------------------------------------
-	//                               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.putShort(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.putChar(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 >>> 0) & 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 >> 0) & 0x3F));
-			} else {
-				bytearr[count++] = (byte) (0xC0 | ((c >> 6) & 0x1F));
-				bytearr[count++] = (byte) (0x80 | ((c >> 0) & 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;
-		}
-		return;
-	}
-
-	@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();
-		}
-		return;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/CheckedMemorySegment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/CheckedMemorySegment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/CheckedMemorySegment.java
deleted file mode 100644
index 85c91c0..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/CheckedMemorySegment.java
+++ /dev/null
@@ -1,407 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.memorymanager;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-/**
- * This class represents a piece of memory allocated from the memory manager. The segment is backed
- * by a byte array and features random put and get methods for the basic types that are stored in a byte-wise
- * fashion in the memory.
- */
-public class CheckedMemorySegment {
-	
-	/**
-	 * The array in which the data is stored.
-	 */
-	protected byte[] memory;
-	
-	/**
-	 * The offset in the memory array where this segment starts.
-	 */
-	protected final int offset;
-	
-	/**
-	 * The size of the memory segment.
-	 */
-	protected final int size;
-	
-	/**
-	 * Wrapper for I/O requests.
-	 */
-	protected ByteBuffer wrapper;
-	
-	// -------------------------------------------------------------------------
-	//                             Constructors
-	// -------------------------------------------------------------------------
-
-	public CheckedMemorySegment(byte[] memory) {
-		this.memory = memory;
-		this.offset = 0;
-		this.size = memory.length;
-	}
-
-	// -------------------------------------------------------------------------
-	//                        MemorySegment Accessors
-	// -------------------------------------------------------------------------
-
-	public boolean isFreed() {
-		return this.memory == null;
-	}
-	
-	public final int size() {
-		return size;
-	}
-
-	public final byte[] getBackingArray() {
-		return this.memory;
-	}
-
-	public final int translateOffset(int offset) {
-		return this.offset + offset;
-	}
-	
-	// -------------------------------------------------------------------------
-	//                       Helper methods
-	// -------------------------------------------------------------------------
-
-	public ByteBuffer wrap(int offset, int length) {
-		if (offset > this.size || offset > this.size - length) {
-			throw new IndexOutOfBoundsException();
-		}
-		
-		if (this.wrapper == null) {
-			this.wrapper = ByteBuffer.wrap(this.memory, this.offset + offset, length);
-		}
-		else {
-			this.wrapper.position(this.offset + offset);
-			this.wrapper.limit(this.offset + offset + length);
-		}
-		
-		return this.wrapper;
-	}
-
-
-	// --------------------------------------------------------------------
-	//                            Random Access
-	// --------------------------------------------------------------------
-
-	// ------------------------------------------------------------------------------------------------------
-	// WARNING: Any code for range checking must take care to avoid integer overflows. The position
-	// integer may go up to <code>Integer.MAX_VALUE</tt>. Range checks that work after the principle
-	// <code>position + 3 &lt; end</code> may fail because <code>position + 3</code> becomes negative.
-	// A safe solution is to subtract the delta from the limit, for example
-	// <code>position &lt; end - 3</code>. Since all indices are always positive, and the integer domain
-	// has one more negative value than positive values, this can never cause an underflow.
-	// ------------------------------------------------------------------------------------------------------
-
-	public final byte get(int index) {
-		if (index >= 0 && index < this.size) {
-			return this.memory[this.offset + index];
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final CheckedMemorySegment put(int index, byte b) {
-		if (index >= 0 && index < this.size) {
-			this.memory[this.offset + index] = b;
-			return this;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final CheckedMemorySegment get(int index, byte[] dst) {
-		return get(index, dst, 0, dst.length);
-	}
-
-	public final CheckedMemorySegment put(int index, byte[] src) {
-		return put(index, src, 0, src.length);
-	}
-
-	public final CheckedMemorySegment get(int index, byte[] dst, int offset, int length) {
-		if (index >= 0 && index < this.size && index <= this.size - length && offset <= dst.length - length) {
-			System.arraycopy(this.memory, this.offset + index, dst, offset, length);
-			return this;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final CheckedMemorySegment put(int index, byte[] src, int offset, int length) {
-		if (index >= 0 && index < this.size && index <= this.size - length && offset <= src.length - length) {
-			System.arraycopy(src, offset, this.memory, this.offset + index, length);
-			return this;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final CheckedMemorySegment get(DataOutput out, int offset, int length) throws IOException {
-		if (offset >= 0 && offset < this.size && length >= 0 && offset <= this.size - length) {
-			out.write(this.memory, this.offset + offset, length);
-			return this;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final CheckedMemorySegment put(DataInput in, int offset, int length) throws IOException {
-		if (offset >= 0 && offset < this.size && length >= 0 && offset <= this.size - length) {
-			in.readFully(this.memory, this.offset + offset, length);
-			return this;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final boolean getBoolean(int index) {
-		if (index >= 0 && index < this.size) {
-			return this.memory[this.offset + index] != 0;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final CheckedMemorySegment putBoolean(int index, boolean value) {
-		if (index >= 0 && index < this.size) {
-			this.memory[this.offset + index] = (byte) (value ? 1 : 0);
-			return this;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final char getChar(int index) {
-		if (index >= 0 && index < this.size - 1) {
-			return (char) ( ((this.memory[this.offset + index + 0] & 0xff) << 8) | 
-							(this.memory[this.offset + index + 1] & 0xff) );
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final CheckedMemorySegment putChar(int index, char value) {
-		if (index >= 0 && index < this.size - 1) {
-			this.memory[this.offset + index + 0] = (byte) (value >> 8);
-			this.memory[this.offset + index + 1] = (byte) value;
-			return this;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final short getShort(int index) {
-		if (index >= 0 && index < this.size - 1) {
-			return (short) (
-					((this.memory[this.offset + index + 0] & 0xff) << 8) |
-					((this.memory[this.offset + index + 1] & 0xff)) );
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final CheckedMemorySegment putShort(int index, short value) {
-		if (index >= 0 && index < this.size - 1) {
-			this.memory[this.offset + index + 0] = (byte) (value >> 8);
-			this.memory[this.offset + index + 1] = (byte) value;
-			return this;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-	
-	public final int getInt(int index) {
-		if (index >= 0 && index < this.size - 3) {
-			return ((this.memory[index    ] & 0xff) << 24)
-				| ((this.memory[index + 1] & 0xff) << 16)
-				| ((this.memory[index + 2] & 0xff) <<  8)
-				| ((this.memory[index + 3] & 0xff)     );
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final int getIntLittleEndian(int index) {
-		if (index >= 0 && index < this.size - 3) {
-			return ((this.memory[index    ] & 0xff)      )
-				| ((this.memory[index + 1] & 0xff) <<  8)
-				| ((this.memory[index + 2] & 0xff) << 16)
-				| ((this.memory[index + 3] & 0xff) << 24);
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-	
-	public final int getIntBigEndian(int index) {
-		if (index >= 0 && index < this.size - 3) {
-			return ((this.memory[index    ] & 0xff) << 24)
-				| ((this.memory[index + 1] & 0xff) << 16)
-				| ((this.memory[index + 2] & 0xff) <<  8)
-				| ((this.memory[index + 3] & 0xff)      );
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-	
-	public final void putInt(int index, int value) {
-		if (index >= 0 && index < this.size - 3) {
-			this.memory[index    ] = (byte) (value >> 24);
-			this.memory[index + 1] = (byte) (value >> 16);
-			this.memory[index + 2] = (byte) (value >> 8);
-			this.memory[index + 3] = (byte) value;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-	
-	public final void putIntLittleEndian(int index, int value) {
-		if (index >= 0 && index < this.size - 3) {
-			this.memory[index    ] = (byte) value;
-			this.memory[index + 1] = (byte) (value >>  8);
-			this.memory[index + 2] = (byte) (value >> 16);
-			this.memory[index + 3] = (byte) (value >> 24);
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-	
-	public final void putIntBigEndian(int index, int value) {
-		if (index >= 0 && index < this.size - 3) {
-			this.memory[index    ] = (byte) (value >> 24);
-			this.memory[index + 1] = (byte) (value >> 16);
-			this.memory[index + 2] = (byte) (value >> 8);
-			this.memory[index + 3] = (byte) value;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-	
-	public final long getLong(int index) {
-		if (index >= 0 && index < this.size - 7) {
-			return (((long) this.memory[index    ] & 0xff) << 56)
-				| (((long) this.memory[index + 1] & 0xff) << 48)
-				| (((long) this.memory[index + 2] & 0xff) << 40)
-				| (((long) this.memory[index + 3] & 0xff) << 32)
-				| (((long) this.memory[index + 4] & 0xff) << 24)
-				| (((long) this.memory[index + 5] & 0xff) << 16)
-				| (((long) this.memory[index + 6] & 0xff) <<  8)
-				| (((long) this.memory[index + 7] & 0xff)      );
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-	public final long getLongLittleEndian(int index) {
-		if (index >= 0 && index < this.size - 7) {
-			return (((long) this.memory[index    ] & 0xff)      )
-				| (((long) this.memory[index + 1] & 0xff) <<  8)
-				| (((long) this.memory[index + 2] & 0xff) << 16)
-				| (((long) this.memory[index + 3] & 0xff) << 24)
-				| (((long) this.memory[index + 4] & 0xff) << 32)
-				| (((long) this.memory[index + 5] & 0xff) << 40)
-				| (((long) this.memory[index + 6] & 0xff) << 48)
-				| (((long) this.memory[index + 7] & 0xff) << 56);
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-	
-	public final long getLongBigEndian(int index) {
-		if (index >= 0 && index < this.size - 7) {
-			return (((long) this.memory[index    ] & 0xff) << 56)
-				| (((long) this.memory[index + 1] & 0xff) << 48)
-				| (((long) this.memory[index + 2] & 0xff) << 40)
-				| (((long) this.memory[index + 3] & 0xff) << 32)
-				| (((long) this.memory[index + 4] & 0xff) << 24)
-				| (((long) this.memory[index + 5] & 0xff) << 16)
-				| (((long) this.memory[index + 6] & 0xff) <<  8)
-				| (((long) this.memory[index + 7] & 0xff)      );
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final void putLong(int index, long value) {
-		if (index >= 0 && index < this.size - 7) {
-			this.memory[index    ] = (byte) (value >> 56);
-			this.memory[index + 1] = (byte) (value >> 48);
-			this.memory[index + 2] = (byte) (value >> 40);
-			this.memory[index + 3] = (byte) (value >> 32);
-			this.memory[index + 4] = (byte) (value >> 24);
-			this.memory[index + 5] = (byte) (value >> 16);
-			this.memory[index + 6] = (byte) (value >>  8);
-			this.memory[index + 7] = (byte)  value;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-	
-	public final void putLongLittleEndian(int index, long value) {
-		if (index >= 0 && index < this.size - 7) {
-			this.memory[index    ] = (byte)  value;
-			this.memory[index + 1] = (byte) (value >>  8);
-			this.memory[index + 2] = (byte) (value >> 16);
-			this.memory[index + 3] = (byte) (value >> 24);
-			this.memory[index + 4] = (byte) (value >> 32);
-			this.memory[index + 5] = (byte) (value >> 40);
-			this.memory[index + 6] = (byte) (value >> 48);
-			this.memory[index + 7] = (byte) (value >> 56);
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-	
-	public final void putLongBigEndian(int index, long value) {
-		if (index >= 0 && index < this.size - 7) {
-			this.memory[index    ] = (byte) (value >> 56);
-			this.memory[index + 1] = (byte) (value >> 48);
-			this.memory[index + 2] = (byte) (value >> 40);
-			this.memory[index + 3] = (byte) (value >> 32);
-			this.memory[index + 4] = (byte) (value >> 24);
-			this.memory[index + 5] = (byte) (value >> 16);
-			this.memory[index + 6] = (byte) (value >>  8);
-			this.memory[index + 7] = (byte)  value;
-		} else {
-			throw new IndexOutOfBoundsException();
-		}
-	}
-
-	public final float getFloat(int index) {
-		return Float.intBitsToFloat(getInt(index));
-	}
-
-	public final CheckedMemorySegment putFloat(int index, float value) {
-		putLong(index, Float.floatToIntBits(value));
-		return this;
-	}
-
-	public final double getDouble(int index) {
-		return Double.longBitsToDouble(getLong(index));
-	}
-
-	public final CheckedMemorySegment putDouble(int index, double value) {
-		putLong(index, Double.doubleToLongBits(value));
-		return this;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/DefaultMemoryManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/DefaultMemoryManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/DefaultMemoryManager.java
deleted file mode 100644
index b041ac9..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/DefaultMemoryManager.java
+++ /dev/null
@@ -1,490 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.memorymanager;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-
-import java.util.ArrayDeque;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.ConcurrentModificationException;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Set;
-
-
-public class DefaultMemoryManager implements MemoryManager {
-	
-	/**
-	 * The default memory page size. Currently set to 32 KiBytes.
-	 */
-	public static final int DEFAULT_PAGE_SIZE = 32 * 1024;
-	
-	/**
-	 * The minimal memory page size. Currently set to 4 KiBytes.
-	 */
-	public static final int MIN_PAGE_SIZE = 4 * 1024;
-	
-	/**
-	 * The Logger.
-	 */
-	private static final Logger LOG = LoggerFactory.getLogger(DefaultMemoryManager.class);
-	
-	// --------------------------------------------------------------------------------------------
-	
-	private final Object lock = new Object();	 	// The lock used on the shared structures.
-	
-	private final ArrayDeque<byte[]> freeSegments;	// the free memory segments
-	
-	private final HashMap<AbstractInvokable, Set<DefaultMemorySegment>> allocatedSegments;
-	
-	private final long roundingMask;		// mask used to round down sizes to multiples of the page size
-	
-	private final int pageSize;				// the page size, in bytes
-	
-	private final int pageSizeBits;			// the number of bits that the power-of-two page size corresponds to
-	
-	private final int totalNumPages;		// The initial total size, for verification.
-
-	/** The total size of the memory managed by this memory manager */
-	private final long memorySize;
-
-	/** Number of slots of the task manager */
-	private final int numberOfSlots;
-	
-	private final boolean isPreAllocated;
-	
-	/** The number of memory pages that have not been allocated and are available for lazy allocation */
-	private int numNonAllocatedPages;
-	
-	/** flag whether the close() has already been invoked */
-	private boolean isShutDown;
-
-	// ------------------------------------------------------------------------
-	// Constructors / Destructors
-	// ------------------------------------------------------------------------
-
-	/**
-	 * Creates a memory manager with the given capacity, using the default page size.
-	 * 
-	 * @param memorySize The total size of the memory to be managed by this memory manager.
-	 * @param numberOfSlots The number of slots of the task manager.
-	 */
-	public DefaultMemoryManager(long memorySize, int numberOfSlots) {
-		this(memorySize, numberOfSlots, DEFAULT_PAGE_SIZE, true);
-	}
-
-	/**
-	 * Creates a memory manager with the given capacity and given page size.
-	 * 
-	 * @param memorySize The total size of the memory to be managed by this memory manager.
-	 * @param numberOfSlots The number of slots of the task manager.
-	 * @param pageSize The size of the pages handed out by the memory manager.
-	 * @param preAllocateMemory True, if the memory manaber should immediately allocate all memory, false
-	 *                          if it should allocate and release the memory as needed.
-	 */
-	public DefaultMemoryManager(long memorySize, int numberOfSlots, int pageSize, boolean preAllocateMemory) {
-		// sanity checks
-		if (memorySize <= 0) {
-			throw new IllegalArgumentException("Size of total memory must be positive.");
-		}
-		if (pageSize < MIN_PAGE_SIZE) {
-			throw new IllegalArgumentException("The page size must be at least " + MIN_PAGE_SIZE + " bytes.");
-		}
-		if ((pageSize & (pageSize - 1)) != 0) {
-			// not a power of two
-			throw new IllegalArgumentException("The given page size is not a power of two.");
-		}
-
-		this.memorySize = memorySize;
-
-		this.numberOfSlots = numberOfSlots;
-		
-		// assign page size and bit utilities
-		this.pageSize = pageSize;
-		this.roundingMask = ~((long) (pageSize - 1));
-		int log = 0;
-		while ((pageSize = pageSize >>> 1) != 0) {
-			log++;
-		}
-		this.pageSizeBits = log;
-		
-		this.totalNumPages = getNumPages(memorySize);
-		if (this.totalNumPages < 1) {
-			throw new IllegalArgumentException("The given amount of memory amounted to less than one page.");
-		}
-		
-		// initialize the free segments and allocated segments tracking structures
-		this.freeSegments = new ArrayDeque<byte[]>(this.totalNumPages);
-		this.allocatedSegments = new HashMap<AbstractInvokable, Set<DefaultMemorySegment>>();
-
-		this.isPreAllocated = preAllocateMemory;
-		
-		if (preAllocateMemory) {
-			// add the full chunks
-			for (int i = 0; i < this.totalNumPages; i++) {
-				// allocate memory of the specified size
-				this.freeSegments.add(new byte[this.pageSize]);
-			}
-		}
-		else {
-			this.numNonAllocatedPages = this.totalNumPages;
-		}
-	}
-
-	@Override
-	public void shutdown() {
-		// -------------------- BEGIN CRITICAL SECTION -------------------
-		synchronized (this.lock)
-		{
-			if (!this.isShutDown) {
-				if (LOG.isDebugEnabled()) {
-					LOG.debug("Shutting down MemoryManager instance " + this);
-				}
-	
-				// mark as shutdown and release memory
-				this.isShutDown = true;
-				
-				this.freeSegments.clear();
-				this.numNonAllocatedPages = 0;
-				
-				// go over all allocated segments and release them
-				for (Set<DefaultMemorySegment> segments : this.allocatedSegments.values()) {
-					for (DefaultMemorySegment seg : segments) {
-						seg.destroy();
-					}
-				}
-			}
-		}
-		// -------------------- END CRITICAL SECTION -------------------
-	}
-
-	@Override
-	public boolean isShutdown() {
-		return this.isShutDown;
-	}
-
-	@Override
-	public boolean verifyEmpty() {
-		synchronized (this.lock) {
-			return isPreAllocated ?
-					this.freeSegments.size() == this.totalNumPages :
-					this.numNonAllocatedPages == this.totalNumPages;
-		}
-	}
-
-	// ------------------------------------------------------------------------
-	//                 MemoryManager interface implementation
-	// ------------------------------------------------------------------------
-
-	@Override
-	public List<MemorySegment> allocatePages(AbstractInvokable owner, int numPages) throws MemoryAllocationException {
-		final ArrayList<MemorySegment> segs = new ArrayList<MemorySegment>(numPages);
-		allocatePages(owner, segs, numPages);
-		return segs;
-	}
-
-	@Override
-	public void allocatePages(AbstractInvokable owner, List<MemorySegment> target, int numPages)
-			throws MemoryAllocationException
-	{
-		// sanity check
-		if (owner == null) {
-			throw new IllegalAccessError("The memory owner must not be null.");
-		}
-		
-		// reserve array space, if applicable
-		if (target instanceof ArrayList) {
-			((ArrayList<MemorySegment>) target).ensureCapacity(numPages);
-		}
-		
-		// -------------------- BEGIN CRITICAL SECTION -------------------
-		synchronized (this.lock)
-		{
-			if (this.isShutDown) {
-				throw new IllegalStateException("Memory manager has been shut down.");
-			}
-			
-			// in the case of pre-allocated memory, the 'numNonAllocatedPages' is zero, in the
-			// lazy case, the 'freeSegments.size()' is zero.
-			if (numPages > (this.freeSegments.size() + numNonAllocatedPages)) {
-				throw new MemoryAllocationException("Could not allocate " + numPages + " pages. Only " + 
-					this.freeSegments.size() + " pages are remaining.");
-			}
-			
-			Set<DefaultMemorySegment> segmentsForOwner = this.allocatedSegments.get(owner);
-			if (segmentsForOwner == null) {
-				segmentsForOwner = new HashSet<DefaultMemorySegment>(4 * numPages / 3 + 1);
-				this.allocatedSegments.put(owner, segmentsForOwner);
-			}
-			
-			if (isPreAllocated) {
-				for (int i = numPages; i > 0; i--) {
-					byte[] buffer = this.freeSegments.poll();
-					final DefaultMemorySegment segment = new DefaultMemorySegment(owner, buffer);
-					target.add(segment);
-					segmentsForOwner.add(segment);
-				}
-			}
-			else {
-				for (int i = numPages; i > 0; i--) {
-					byte[] buffer = new byte[pageSize];
-					final DefaultMemorySegment segment = new DefaultMemorySegment(owner, buffer);
-					target.add(segment);
-					segmentsForOwner.add(segment);
-				}
-				numNonAllocatedPages -= numPages;
-			}
-		}
-		// -------------------- END CRITICAL SECTION -------------------
-	}
-	
-	// ------------------------------------------------------------------------
-
-	@Override
-	public void release(MemorySegment segment) {
-		// check if segment is null or has already been freed
-		if (segment == null || segment.isFreed() || !(segment instanceof DefaultMemorySegment)) {
-			return;
-		}
-		
-		final DefaultMemorySegment defSeg = (DefaultMemorySegment) segment;
-		final AbstractInvokable owner = defSeg.owner;
-		
-		// -------------------- BEGIN CRITICAL SECTION -------------------
-		synchronized (this.lock)
-		{
-			if (this.isShutDown) {
-				throw new IllegalStateException("Memory manager has been shut down.");
-			}
-
-			// remove the reference in the map for the owner
-			try {
-				Set<DefaultMemorySegment> segsForOwner = this.allocatedSegments.get(owner);
-				
-				if (segsForOwner != null) {
-					segsForOwner.remove(defSeg);
-					if (segsForOwner.isEmpty()) {
-						this.allocatedSegments.remove(owner);
-					}
-				}
-
-				byte[] buffer = defSeg.destroy();
-				
-				if (isPreAllocated) {
-					// release the memory in any case
-					this.freeSegments.add(buffer);
-				}
-				else {
-					numNonAllocatedPages++;
-				}
-			}
-			catch (Throwable t) {
-				throw new RuntimeException("Error removing book-keeping reference to allocated memory segment.", t);
-			}
-		}
-		// -------------------- END CRITICAL SECTION -------------------
-	}
-
-	@Override
-	public <T extends MemorySegment> void release(Collection<T> segments) {
-		if (segments == null) {
-			return;
-		}
-		
-		// -------------------- BEGIN CRITICAL SECTION -------------------
-		synchronized (this.lock)
-		{
-			if (this.isShutDown) {
-				throw new IllegalStateException("Memory manager has been shut down.");
-			}
-
-			// since concurrent modifications to the collection
-			// can disturb the release, we need to try potentially multiple times
-			boolean successfullyReleased = false;
-			do {
-				final Iterator<T> segmentsIterator = segments.iterator();
-
-				AbstractInvokable lastOwner = null;
-				Set<DefaultMemorySegment> segsForOwner = null;
-
-				try {
-					// go over all segments
-					while (segmentsIterator.hasNext()) {
-
-						final MemorySegment seg = segmentsIterator.next();
-						if (seg == null || seg.isFreed()) {
-							continue;
-						}
-
-						final DefaultMemorySegment defSeg = (DefaultMemorySegment) seg;
-						final AbstractInvokable owner = defSeg.owner;
-
-						try {
-							// get the list of segments by this owner only if it is a different owner than for
-							// the previous one (or it is the first segment)
-							if (lastOwner != owner) {
-								lastOwner = owner;
-								segsForOwner = this.allocatedSegments.get(owner);
-							}
-
-							// remove the segment from the list
-							if (segsForOwner != null) {
-								segsForOwner.remove(defSeg);
-								if (segsForOwner.isEmpty()) {
-									this.allocatedSegments.remove(owner);
-								}
-							}
-
-							// release the memory in any case
-							byte[] buffer = defSeg.destroy();
-							
-							if (isPreAllocated) {
-								this.freeSegments.add(buffer);
-							}
-							else {
-								numNonAllocatedPages++;
-							}
-						}
-						catch (Throwable t) {
-							throw new RuntimeException(
-									"Error removing book-keeping reference to allocated memory segment.", t);
-						}
-					}
-
-					segments.clear();
-
-					// the only way to exit the loop
-					successfullyReleased = true;
-				}
-				catch (ConcurrentModificationException e) {
-					// this may happen in the case where an asynchronous
-					// call releases the memory. fall through the loop and try again
-				}
-			} while (!successfullyReleased);
-		}
-		// -------------------- END CRITICAL SECTION -------------------
-	}
-
-	@Override
-	public void releaseAll(AbstractInvokable owner) {
-		// -------------------- BEGIN CRITICAL SECTION -------------------
-		synchronized (this.lock)
-		{
-			if (this.isShutDown) {
-				throw new IllegalStateException("Memory manager has been shut down.");
-			}
-
-			// get all segments
-			final Set<DefaultMemorySegment> segments = this.allocatedSegments.remove(owner);
-
-			// all segments may have been freed previously individually
-			if (segments == null || segments.isEmpty()) {
-				return;
-			}
-
-			// free each segment
-			if (isPreAllocated) {
-				for (DefaultMemorySegment seg : segments) {
-					final byte[] buffer = seg.destroy();
-					this.freeSegments.add(buffer);
-				}
-			}
-			else {
-				for (DefaultMemorySegment seg : segments) {
-					seg.destroy();
-				}
-				numNonAllocatedPages += segments.size();
-			}
-
-			segments.clear();
-		}
-		// -------------------- END CRITICAL SECTION -------------------
-	}
-	
-	// ------------------------------------------------------------------------
-
-	@Override
-	public int getPageSize() {
-		return this.pageSize;
-	}
-
-	@Override
-	public long getMemorySize() {
-		return this.memorySize;
-	}
-
-	@Override
-	public int computeNumberOfPages(double fraction) {
-		if (fraction <= 0 || fraction > 1) {
-			throw new IllegalArgumentException("The fraction of memory to allocate must within (0, 1].");
-		}
-
-		return (int)(this.totalNumPages * fraction / this.numberOfSlots);
-	}
-
-	@Override
-	public long computeMemorySize(double fraction) {
-		return this.pageSize * computeNumberOfPages(fraction);
-	}
-
-	@Override
-	public long roundDownToPageSizeMultiple(long numBytes) {
-		return numBytes & this.roundingMask;
-	}
-	
-	// ------------------------------------------------------------------------
-	
-	private int getNumPages(long numBytes) {
-		if (numBytes < 0) {
-			throw new IllegalArgumentException("The number of bytes to allocate must not be negative.");
-		}
-		
-		final long numPages = numBytes >>> this.pageSizeBits;
-		if (numPages <= Integer.MAX_VALUE) {
-			return (int) numPages;
-		} else {
-			throw new IllegalArgumentException("The given number of bytes corresponds to more than MAX_INT pages.");
-		}
-	}
-	
-	// ------------------------------------------------------------------------
-	
-	private static final class DefaultMemorySegment extends MemorySegment {
-		
-		private AbstractInvokable owner;
-		
-		DefaultMemorySegment(AbstractInvokable owner, byte[] memory) {
-			super(memory);
-			this.owner = owner;
-		}
-		
-		byte[] destroy() {
-			final byte[] buffer = this.memory;
-			this.memory = null;
-			this.wrapper = null;
-			return buffer;
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/ListMemorySegmentSource.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/ListMemorySegmentSource.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/ListMemorySegmentSource.java
deleted file mode 100644
index cacb6b0..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/ListMemorySegmentSource.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.memorymanager;
-
-import java.util.List;
-
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.core.memory.MemorySegmentSource;
-
-/**
- * Simple memory segment source that draws segments from a list.
- * 
- */
-public class ListMemorySegmentSource implements MemorySegmentSource
-{
-	private final List<MemorySegment> segments;
-	
-	public ListMemorySegmentSource(final List<MemorySegment> memorySegments) {
-		this.segments = memorySegments;
-	}
-	
-
-	@Override
-	public MemorySegment nextSegment() {
-		if (this.segments.size() > 0) {
-			return this.segments.remove(this.segments.size() - 1);
-		} else {
-			return null;
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/MemoryAllocationException.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/MemoryAllocationException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/MemoryAllocationException.java
deleted file mode 100644
index d4d8b3b..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/MemoryAllocationException.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.memorymanager;
-
-/**
- * An exception to be thrown when a memory allocation operation is not successful.
- */
-public class MemoryAllocationException extends Exception {
-	
-	private static final long serialVersionUID = -403983866457947012L;
-
-	public MemoryAllocationException() {
-		super();
-	}
-
-	public MemoryAllocationException(String message, Throwable cause) {
-		super(message, cause);
-	}
-
-	public MemoryAllocationException(String message) {
-		super(message);
-	}
-
-	public MemoryAllocationException(Throwable cause) {
-		super(cause);
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/MemoryManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/MemoryManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/MemoryManager.java
deleted file mode 100644
index 631f0b8..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/MemoryManager.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.memorymanager;
-
-import java.util.Collection;
-import java.util.List;
-
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-
-/**
- * Interface for a memory manager that assigns portions of memory to different tasks. Each allocated segment of
- * memory is specific to a task. Memory segments can be freed individually, or all memory allocated by a task can 
- * be freed as a whole.
- * <p>
- * Internally, memory is represented as byte arrays. The memory manager acts like a distributer for memory, which
- * means it assigns portions of the arrays to tasks. If memory is released, it means that this part of the memory can
- * be assigned to other tasks.
- */
-public interface MemoryManager {
-	
-	List<MemorySegment> allocatePages(AbstractInvokable owner, int numPages) throws MemoryAllocationException;
-	
-	void allocatePages(AbstractInvokable owner, List<MemorySegment> target, int numPages) throws MemoryAllocationException;
-	
-	/**
-	 * Tries to release the memory for the specified segment. If the <code>segment</code> has already been released or
-	 * is <code>null</code>, the request is simply ignored. If the segment is not from the expected
-	 * MemorySegment implementation type, an <code>IllegalArgumentException</code> is thrown.
-	 * 
-	 * @param segment The segment to be released.
-	 * @throws IllegalArgumentException Thrown, if the given segment is of an incompatible type.
-	 */
-	void release(MemorySegment segment);
-
-	/**
-	 * Tries to release the memory for the specified collection of segments.
-	 * 
-	 * @param <T> The type of memory segment.
-	 * @param segments The segments to be released.
-	 * @throws NullPointerException Thrown, if the given collection is null.
-	 * @throws IllegalArgumentException Thrown, id the segments are of an incompatible type.
-	 */
-	<T extends MemorySegment> void release(Collection<T> segments);
-	
-	/**
-	 * Releases all memory segments for the given task. 
-	 *
-	 * @param task The task whose memory segments are to be released.
-	 */
-	void releaseAll(AbstractInvokable task);
-	
-	// --------------------------------------------------------------------------------------------
-	
-	/**
-	 * Gets the size of the pages handled by the memory manager.
-	 * 
-	 * @return The size of the pages handled by the memory manager.
-	 */
-	int getPageSize();
-
-	/**
-	 * Returns the total size of memory.
-	 * 
-	 * @return The total size of memory.
-	 */
-	long getMemorySize();
-	
-	/**
-	 * Computes to how many pages the given number of bytes corresponds. If the given number of bytes is not an
-	 * exact multiple of a page size, the result is rounded down, such that a portion of the memory (smaller
-	 * than the page size) is not included.
-	 * 
-	 * @param fraction the fraction of the total memory per slot
-	 * @return The number of pages to which 
-	 */
-	int computeNumberOfPages(double fraction);
-
-	/**
-	 * Computes the memory size of the fraction per slot.
-	 * 
-	 * @param fraction The fraction of the memory of the task slot.
-	 * @return The number of pages corresponding to the memory fraction.
-	 */
-	long computeMemorySize(double fraction);
-	
-	/**
-	 * Rounds the given value down to a multiple of the memory manager's page size.
-	 * 
-	 * @return The given value, rounded down to a multiple of the page size.
-	 */
-	long roundDownToPageSizeMultiple(long numBytes);
-
-	// --------------------------------------------------------------------------------------------
-	
-	/**
-	 * Shuts the memory manager down, trying to release all the memory it managed. Depending
-	 * on implementation details, the memory does not necessarily become reclaimable by the
-	 * garbage collector, because there might still be references to allocated segments in the
-	 * code that allocated them from the memory manager.
-	 */
-	void shutdown();
-
-	/**
-	 * Checks whether the MemoryManager has been shut down.
-	 *
-	 * @return True, if the memory manager is shut down, false otherwise.
-	 */
-	boolean isShutdown();
-
-	/**
-	 * Checks if the memory manager all memory available.
-	 * 
-	 * @return True, if the memory manager is empty and valid, false if it is not empty or corrupted.
-	 */
-	boolean verifyEmpty();
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/655a891d/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/SimpleMemorySegment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/SimpleMemorySegment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/SimpleMemorySegment.java
deleted file mode 100644
index 7219503..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memorymanager/SimpleMemorySegment.java
+++ /dev/null
@@ -1,329 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.memorymanager;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-/**
- * This class represents a piece of memory allocated from the memory manager. The segment is backed
- * by a byte array and features random put and get methods for the basic types that are stored in a byte-wise
- * fashion in the memory.
- */
-public class SimpleMemorySegment {
-	
-	/**
-	 * The array in which the data is stored.
-	 */
-	protected byte[] memory;
-	
-	/**
-	 * Wrapper for I/O requests.
-	 */
-	protected ByteBuffer wrapper;
-	
-	// -------------------------------------------------------------------------
-	//                             Constructors
-	// -------------------------------------------------------------------------
-
-	/**
-	 * Creates a new memory segment of given size with the provided views.
-	 */
-	public SimpleMemorySegment(byte[] memory) {
-		this.memory = memory;
-	}
-
-	// -------------------------------------------------------------------------
-	//                        MemorySegment Accessors
-	// -------------------------------------------------------------------------
-	
-	/**
-	 * Checks whether this memory segment has already been freed. In that case, the
-	 * segment must not be used any more.
-	 * 
-	 * @return True, if the segment has been freed, false otherwise.
-	 */
-	public final boolean isFreed() {
-		return this.memory == null;
-	}
-	
-	/**
-	 * Gets the size of the memory segment, in bytes. Because segments
-	 * are backed by arrays, they cannot be larger than two GiBytes.
-	 * 
-	 * @return The size in bytes.
-	 */
-	public final int size() {
-		return this.memory.length;
-	}
-	
-	/**
-	 * Gets the byte array that backs the memory segment and this random access view.
-	 * Since different regions of the backing array are used by different segments, the logical
-	 * positions in this view do not correspond to the indexes in the backing array and need
-	 * to be translated via the {@link #translateOffset(int)} method.
-	 * 
-	 * @return The backing byte array.
-	 */
-	public final byte[] getBackingArray() {
-		return this.memory;
-	}
-
-	/**
-	 * Translates the given offset for this view into the offset for the backing array.
-	 * 
-	 * @param offset The offset to be translated.
-	 * @return The corresponding position in the backing array.
-	 */
-	public final int translateOffset(int offset) {
-		return offset;
-	}
-	
-	// -------------------------------------------------------------------------
-	//                       Helper methods
-	// -------------------------------------------------------------------------
-	
-
-	/**
-	 * Wraps the chunk of the underlying memory located between <tt>offset<tt> and 
-	 * <tt>length</tt> in a NIO ByteBuffer.
-	 * 
-	 * @param offset The offset in the memory segment.
-	 * @param length The number of bytes to be wrapped as a buffer.
-	 * @return A <tt>ByteBuffer</tt> backed by the specified portion of the memory segment.
-	 * @throws IndexOutOfBoundsException Thrown, if offset is negative or larger than the memory segment size,
-	 *                                   or if the offset plus the length is larger than the segment size.
-	 */
-	public 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.position(offset);
-			this.wrapper.limit(offset + length);
-		}
-		
-		return this.wrapper;
-	}
-
-
-	// --------------------------------------------------------------------
-	//                            Random Access
-	// --------------------------------------------------------------------
-
-	// ------------------------------------------------------------------------------------------------------
-	// WARNING: Any code for range checking must take care to avoid integer overflows. The position
-	// integer may go up to <code>Integer.MAX_VALUE</tt>. Range checks that work after the principle
-	// <code>position + 3 &lt; end</code> may fail because <code>position + 3</code> becomes negative.
-	// A safe solution is to subtract the delta from the limit, for example
-	// <code>position &lt; end - 3</code>. Since all indices are always positive, and the integer domain
-	// has one more negative value than positive values, this can never cause an underflow.
-	// ------------------------------------------------------------------------------------------------------
-
-	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(this.memory, index, dst, offset, length);
-	}
-
-	public final void put(int index, byte[] src, int offset, int length) {
-		System.arraycopy(src, offset, this.memory, index, length);
-	}
-
-	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 boolean getBoolean(int index) {
-		return this.memory[index] != 0;
-	}
-
-	public final void putBoolean(int index, boolean value) {
-		this.memory[index] = (byte) (value ? 1 : 0);
-	}
-
-	public final char getChar(int index) {
-		return (char) ( ((this.memory[index    ] & 0xff) << 8) | 
-						(this.memory[index + 1] & 0xff) );
-	}
-
-	public final void putChar(int index, char value) {
-		this.memory[index    ] = (byte) (value >> 8);
-		this.memory[index + 1] = (byte) value;
-	}
-
-	public final short getShort(int index) {
-		return (short) (
-				((this.memory[index    ] & 0xff) << 8) |
-				((this.memory[index + 1] & 0xff)) );
-	}
-
-	public final void putShort(int index, short value) {
-		this.memory[index    ] = (byte) (value >> 8);
-		this.memory[index + 1] = (byte) value;
-	}
-	
-	public final int getInt(int index) {
-		return ((this.memory[index    ] & 0xff) << 24)
-			| ((this.memory[index + 1] & 0xff) << 16)
-			| ((this.memory[index + 2] & 0xff) <<  8)
-			| ((this.memory[index + 3] & 0xff)     );
-	}
-
-	public final int getIntLittleEndian(int index) {
-		return ((this.memory[index    ] & 0xff)      )
-			| ((this.memory[index + 1] & 0xff) <<  8)
-			| ((this.memory[index + 2] & 0xff) << 16)
-			| ((this.memory[index + 3] & 0xff) << 24);
-	}
-	
-	public final int getIntBigEndian(int index) {
-		return ((this.memory[index    ] & 0xff) << 24)
-			| ((this.memory[index + 1] & 0xff) << 16)
-			| ((this.memory[index + 2] & 0xff) <<  8)
-			| ((this.memory[index + 3] & 0xff)      );
-	}
-	
-	public final void putInt(int index, int value) {
-		this.memory[index    ] = (byte) (value >> 24);
-		this.memory[index + 1] = (byte) (value >> 16);
-		this.memory[index + 2] = (byte) (value >> 8);
-		this.memory[index + 3] = (byte) value;
-	}
-	
-	public final void putIntLittleEndian(int index, int value) {
-		this.memory[index    ] = (byte) value;
-		this.memory[index + 1] = (byte) (value >>  8);
-		this.memory[index + 2] = (byte) (value >> 16);
-		this.memory[index + 3] = (byte) (value >> 24);
-	}
-	
-	public final void putIntBigEndian(int index, int value) {
-		this.memory[index    ] = (byte) (value >> 24);
-		this.memory[index + 1] = (byte) (value >> 16);
-		this.memory[index + 2] = (byte) (value >> 8);
-		this.memory[index + 3] = (byte) value;
-	}
-	
-	public final long getLong(int index) {
-		return (((long) this.memory[index    ] & 0xff) << 56)
-			| (((long) this.memory[index + 1] & 0xff) << 48)
-			| (((long) this.memory[index + 2] & 0xff) << 40)
-			| (((long) this.memory[index + 3] & 0xff) << 32)
-			| (((long) this.memory[index + 4] & 0xff) << 24)
-			| (((long) this.memory[index + 5] & 0xff) << 16)
-			| (((long) this.memory[index + 6] & 0xff) <<  8)
-			| (((long) this.memory[index + 7] & 0xff)      );
-	}
-	public final long getLongLittleEndian(int index) {
-		return (((long) this.memory[index    ] & 0xff)      )
-			| (((long) this.memory[index + 1] & 0xff) <<  8)
-			| (((long) this.memory[index + 2] & 0xff) << 16)
-			| (((long) this.memory[index + 3] & 0xff) << 24)
-			| (((long) this.memory[index + 4] & 0xff) << 32)
-			| (((long) this.memory[index + 5] & 0xff) << 40)
-			| (((long) this.memory[index + 6] & 0xff) << 48)
-			| (((long) this.memory[index + 7] & 0xff) << 56);
-	}
-	
-	public final long getLongBigEndian(int index) {
-		return (((long) this.memory[index    ] & 0xff) << 56)
-			| (((long) this.memory[index + 1] & 0xff) << 48)
-			| (((long) this.memory[index + 2] & 0xff) << 40)
-			| (((long) this.memory[index + 3] & 0xff) << 32)
-			| (((long) this.memory[index + 4] & 0xff) << 24)
-			| (((long) this.memory[index + 5] & 0xff) << 16)
-			| (((long) this.memory[index + 6] & 0xff) <<  8)
-			| (((long) this.memory[index + 7] & 0xff)      );
-	}
-
-	public final void putLong(int index, long value) {
-		this.memory[index    ] = (byte) (value >> 56);
-		this.memory[index + 1] = (byte) (value >> 48);
-		this.memory[index + 2] = (byte) (value >> 40);
-		this.memory[index + 3] = (byte) (value >> 32);
-		this.memory[index + 4] = (byte) (value >> 24);
-		this.memory[index + 5] = (byte) (value >> 16);
-		this.memory[index + 6] = (byte) (value >>  8);
-		this.memory[index + 7] = (byte)  value;
-	}
-	
-	public final void putLongLittleEndian(int index, long value) {
-		this.memory[index    ] = (byte)  value;
-		this.memory[index + 1] = (byte) (value >>  8);
-		this.memory[index + 2] = (byte) (value >> 16);
-		this.memory[index + 3] = (byte) (value >> 24);
-		this.memory[index + 4] = (byte) (value >> 32);
-		this.memory[index + 5] = (byte) (value >> 40);
-		this.memory[index + 6] = (byte) (value >> 48);
-		this.memory[index + 7] = (byte) (value >> 56);
-	}
-	
-	public final void putLongBigEndian(int index, long value) {
-		this.memory[index    ] = (byte) (value >> 56);
-		this.memory[index + 1] = (byte) (value >> 48);
-		this.memory[index + 2] = (byte) (value >> 40);
-		this.memory[index + 3] = (byte) (value >> 32);
-		this.memory[index + 4] = (byte) (value >> 24);
-		this.memory[index + 5] = (byte) (value >> 16);
-		this.memory[index + 6] = (byte) (value >>  8);
-		this.memory[index + 7] = (byte)  value;
-	}
-	
-	public final float getFloat(int index) {
-		return Float.intBitsToFloat(getInt(index));
-	}
-
-	public final void putFloat(int index, float value) {
-		putLong(index, Float.floatToIntBits(value));
-	}
-	
-	public final double getDouble(int index) {
-		return Double.longBitsToDouble(getLong(index));
-	}
-
-	public final void putDouble(int index, double value) {
-		putLong(index, Double.doubleToLongBits(value));
-	}
-}