You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2017/10/08 19:38:11 UTC
[01/10] flink git commit: [hotfix] [docs] Fix broken links
Repository: flink
Updated Branches:
refs/heads/master 77e3701ca -> e14caef37
[hotfix] [docs] Fix broken links
This closes #4755
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9e128401
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9e128401
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9e128401
Branch: refs/heads/master
Commit: 9e128401c561345e80b5682c75337802ae44202d
Parents: edc53a7
Author: Gary Yao <gy...@garyyao.com>
Authored: Sun Oct 1 14:11:17 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 20:12:51 2017 +0200
----------------------------------------------------------------------
docs/dev/connectors/index.md | 2 +-
docs/dev/table/common.md | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9e128401/docs/dev/connectors/index.md
----------------------------------------------------------------------
diff --git a/docs/dev/connectors/index.md b/docs/dev/connectors/index.md
index f3ae039..28ab2d3 100644
--- a/docs/dev/connectors/index.md
+++ b/docs/dev/connectors/index.md
@@ -31,7 +31,7 @@ under the License.
## Predefined Sources and Sinks
A few basic data sources and sinks are built into Flink and are always available.
-The [predefined data sources]({{ site.baseurll }}/dev/datastream_api.html#data-sources) include reading from files, directories, and sockets, and
+The [predefined data sources]({{ site.baseurl }}/dev/datastream_api.html#data-sources) include reading from files, directories, and sockets, and
ingesting data from collections and iterators.
The [predefined data sinks]({{ site.baseurl }}/dev/datastream_api.html#data-sinks) support writing to files, to stdout and stderr, and to sockets.
http://git-wip-us.apache.org/repos/asf/flink/blob/9e128401/docs/dev/table/common.md
----------------------------------------------------------------------
diff --git a/docs/dev/table/common.md b/docs/dev/table/common.md
index acd5711..a63c4dc 100644
--- a/docs/dev/table/common.md
+++ b/docs/dev/table/common.md
@@ -332,7 +332,7 @@ The Table API is a language-integrated query API for Scala and Java. In contrast
The API is based on the `Table` class which represents a table (streaming or batch) and offers methods to apply relational operations. These methods return a new `Table` object, which represents the result of applying the relational operation on the input `Table`. Some relational operations are composed of multiple method calls such as `table.groupBy(...).select()`, where `groupBy(...)` specifies a grouping of `table`, and `select(...)` the projection on the grouping of `table`.
-The [Table API]({{ site.baseurl }}/dev/table/tableapi.html) document describes all Table API operations that are supported on streaming and batch tables.
+The [Table API]({{ site.baseurl }}/dev/table/tableApi.html) document describes all Table API operations that are supported on streaming and batch tables.
The following example shows a simple Table API aggregation query:
[09/10] flink git commit: [FLINK-7312] [checkstyle] Enable checkstyle
for 'flink/core/memory/*'
Posted by se...@apache.org.
[FLINK-7312] [checkstyle] Enable checkstyle for 'flink/core/memory/*'
We deliberately ignore redundant modifiers for now since we want `final`
modifiers on `final` classes for increased future-proofness.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c3235c39
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c3235c39
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c3235c39
Branch: refs/heads/master
Commit: c3235c395f7bb69b482b85c6832d427100130ca3
Parents: 9e12840
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Mon Jul 31 12:06:14 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 20:38:02 2017 +0200
----------------------------------------------------------------------
.../apache/flink/core/memory/DataInputView.java | 14 +-
.../core/memory/DataInputViewStreamWrapper.java | 2 +-
.../flink/core/memory/DataOutputView.java | 20 +-
.../memory/DataOutputViewStreamWrapper.java | 4 +-
.../flink/core/memory/HeapMemorySegment.java | 28 +--
.../flink/core/memory/HybridMemorySegment.java | 63 +++---
.../apache/flink/core/memory/MemorySegment.java | 213 +++++++++++--------
.../flink/core/memory/MemorySegmentFactory.java | 45 ++--
.../flink/core/memory/MemorySegmentSource.java | 6 +-
.../apache/flink/core/memory/MemoryUtils.java | 6 +-
.../core/memory/SeekableDataInputView.java | 7 +-
.../core/memory/SeekableDataOutputView.java | 7 +-
.../memory/ByteArrayInputStreamWithPosTest.java | 11 +-
.../ByteArrayOutputStreamWithPosTest.java | 14 +-
.../flink/core/memory/CrossSegmentTypeTest.java | 49 +++--
.../core/memory/EndiannessAccessChecks.java | 19 +-
.../core/memory/HeapMemorySegmentTest.java | 9 +-
.../memory/HybridOffHeapMemorySegmentTest.java | 4 +
.../memory/HybridOnHeapMemorySegmentTest.java | 6 +-
.../core/memory/MemorySegmentChecksTest.java | 14 +-
.../core/memory/MemorySegmentFactoryTest.java | 7 +-
.../core/memory/MemorySegmentTestBase.java | 181 ++++++++--------
.../memory/MemorySegmentUndersizedTest.java | 37 ++--
.../memory/OperationsOnFreedSegmentTest.java | 84 ++++----
.../flink/runtime/memory/MemoryManager.java | 2 +-
tools/maven/suppressions-core.xml | 8 -
26 files changed, 461 insertions(+), 399 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java b/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java
index 88d54d9..7da3fc3 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java
@@ -16,10 +16,8 @@
* limitations under the License.
*/
-
package org.apache.flink.core.memory;
-
import org.apache.flink.annotation.Public;
import java.io.DataInput;
@@ -32,17 +30,17 @@ import java.io.IOException;
*/
@Public
public interface DataInputView extends DataInput {
-
+
/**
* Skips {@code numBytes} bytes of memory. In contrast to the {@link #skipBytes(int)} method,
* this method always skips the desired number of bytes or throws an {@link java.io.EOFException}.
- *
+ *
* @param numBytes The number of bytes to skip.
- *
+ *
* @throws IOException Thrown, if any I/O related problem occurred such that the input could not
* be advanced to the desired position.
*/
- public void skipBytesToRead(int numBytes) throws IOException;
+ void skipBytesToRead(int numBytes) throws IOException;
/**
* Reads up to {@code len} bytes of memory and stores it into {@code b} starting at offset {@code off}.
@@ -54,7 +52,7 @@ public interface DataInputView extends DataInput {
* @return the number of actually read bytes of -1 if there is no more data left
* @throws IOException
*/
- public int read(byte[] b, int off, int len) throws IOException;
+ int read(byte[] b, int off, int len) throws IOException;
/**
* Tries to fill the given byte array {@code b}. Returns the actually number of read bytes or -1 if there is no
@@ -64,5 +62,5 @@ public interface DataInputView extends DataInput {
* @return the number of read bytes or -1 if there is no more data left
* @throws IOException
*/
- public int read(byte[] b) throws IOException;
+ int read(byte[] b) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/DataInputViewStreamWrapper.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/DataInputViewStreamWrapper.java b/flink-core/src/main/java/org/apache/flink/core/memory/DataInputViewStreamWrapper.java
index 19eb285..664dc01 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/DataInputViewStreamWrapper.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/DataInputViewStreamWrapper.java
@@ -34,7 +34,7 @@ public class DataInputViewStreamWrapper extends DataInputStream implements DataI
public DataInputViewStreamWrapper(InputStream in) {
super(in);
}
-
+
@Override
public void skipBytesToRead(int numBytes) throws IOException {
if (skipBytes(numBytes) != numBytes){
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputView.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputView.java b/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputView.java
index d9219b8..3896299 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputView.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputView.java
@@ -16,10 +16,8 @@
* limitations under the License.
*/
-
package org.apache.flink.core.memory;
-
import org.apache.flink.annotation.Public;
import java.io.DataOutput;
@@ -32,26 +30,26 @@ import java.io.IOException;
*/
@Public
public interface DataOutputView extends DataOutput {
-
+
/**
* Skips {@code numBytes} bytes memory. If some program reads the memory that was skipped over, the
- * results are undefined.
- *
+ * results are undefined.
+ *
* @param numBytes The number of bytes to skip.
- *
+ *
* @throws IOException Thrown, if any I/O related problem occurred such that the view could not
* be advanced to the desired position.
*/
- public void skipBytesToWrite(int numBytes) throws IOException;
-
+ void skipBytesToWrite(int numBytes) throws IOException;
+
/**
* Copies {@code numBytes} bytes from the source to this view.
- *
+ *
* @param source The source to copy the bytes from.
* @param numBytes The number of bytes to copy.
- *
+ *
* @throws IOException Thrown, if any I/O related problem occurred, such that either the input view
* could not be read, or the output could not be written.
*/
- public void write(DataInputView source, int numBytes) throws IOException;
+ void write(DataInputView source, int numBytes) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputViewStreamWrapper.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputViewStreamWrapper.java b/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputViewStreamWrapper.java
index 4e45532..5fee01d 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputViewStreamWrapper.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputViewStreamWrapper.java
@@ -31,7 +31,7 @@ import java.io.OutputStream;
public class DataOutputViewStreamWrapper extends DataOutputStream implements DataOutputView {
private byte[] tempBuffer;
-
+
public DataOutputViewStreamWrapper(OutputStream out) {
super(out);
}
@@ -54,7 +54,7 @@ public class DataOutputViewStreamWrapper extends DataOutputStream implements Dat
if (tempBuffer == null) {
tempBuffer = new byte[4096];
}
-
+
while (numBytes > 0) {
int toCopy = Math.min(numBytes, tempBuffer.length);
source.readFully(tempBuffer, 0, toCopy);
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java b/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
index d136404..9aa81d0 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
@@ -30,18 +30,20 @@ import java.util.Objects;
* This class represents a piece of heap memory managed by Flink.
* The segment is backed by a byte array and features random put and get methods for the basic types,
* as well as compare and swap methods.
- * <p>
- * This class specialized byte access and byte copy calls for heap memory, while reusing the
+ *
+ * <p>This class specializes byte access and byte copy calls for heap memory, while reusing the
* multi-byte type accesses and cross-segment operations from the MemorySegment.
- * <p>
- * Note that memory segments should usually not be allocated manually, but rather through the
+ *
+ * <p>Note that memory segments should usually not be allocated manually, but rather through the
* {@link MemorySegmentFactory}.
*/
@Internal
public final class HeapMemorySegment extends MemorySegment {
- /** An extra reference to the heap memory, so we can let byte array checks fail
- * by the built-in checks automatically without extra checks */
+ /**
+ * An extra reference to the heap memory, so we can let byte array checks fail by the built-in
+ * checks automatically without extra checks.
+ */
private byte[] memory;
/**
@@ -53,7 +55,7 @@ public final class HeapMemorySegment extends MemorySegment {
HeapMemorySegment(byte[] memory) {
this(memory, null);
}
-
+
/**
* Creates a new memory segment that represents the data in the given byte array.
* The memory segment references the given owner.
@@ -65,7 +67,7 @@ public final class HeapMemorySegment extends MemorySegment {
super(Objects.requireNonNull(memory), owner);
this.memory = memory;
}
-
+
// -------------------------------------------------------------------------
// MemorySegment operations
// -------------------------------------------------------------------------
@@ -94,11 +96,11 @@ public final class HeapMemorySegment extends MemorySegment {
public byte[] getArray() {
return this.heapMemory;
}
-
+
// ------------------------------------------------------------------------
// Random Access get() and put() methods
// ------------------------------------------------------------------------
-
+
@Override
public final byte get(int index) {
return this.memory[index];
@@ -198,9 +200,11 @@ public final class HeapMemorySegment extends MemorySegment {
"The MemorySegment factory was not initialized for off-heap memory.");
}
- /** prevent external instantiation */
+ /**
+ * Prevent external instantiation.
+ */
HeapMemorySegmentFactory() {}
- };
+ }
public static final HeapMemorySegmentFactory FACTORY = new HeapMemorySegmentFactory();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java b/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
index 1fe37ee..0b23ef3 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
@@ -31,29 +31,32 @@ import java.nio.ByteBuffer;
/**
* This class represents a piece of memory managed by Flink. The memory can be on-heap or off-heap,
* this is transparently handled by this class.
- * <p>
- * This class specialized byte access and byte copy calls for heap memory, while reusing the
+ *
+ * <p>This class specializes byte access and byte copy calls for heap memory, while reusing the
* multi-byte type accesses and cross-segment operations from the MemorySegment.
- * <p>
- * This class subsumes the functionality of the {@link org.apache.flink.core.memory.HeapMemorySegment},
+ *
+ * <p>This class subsumes the functionality of the {@link org.apache.flink.core.memory.HeapMemorySegment},
* but is a bit less efficient for operations on individual bytes.
- * <p>
- * Note that memory segments should usually not be allocated manually, but rather through the
+ *
+ * <p>Note that memory segments should usually not be allocated manually, but rather through the
* {@link MemorySegmentFactory}.
*/
@Internal
public final class HybridMemorySegment extends MemorySegment {
-
- /** The direct byte buffer that allocated the off-heap memory. This memory segment holds a reference
- * to that buffer, so as long as this memory segment lives, the memory will not be released. */
+
+ /**
+ * The direct byte buffer that allocated the off-heap memory. This memory segment holds a
+ * reference to that buffer, so as long as this memory segment lives, the memory will not be
+ * released.
+ */
private final ByteBuffer offHeapBuffer;
/**
* Creates a new memory segment that represents the memory backing the given direct byte buffer.
* Note that the given ByteBuffer must be direct {@link java.nio.ByteBuffer#allocateDirect(int)},
* otherwise this method with throw an IllegalArgumentException.
- * <p>
- * The owner referenced by this memory segment is null.
+ *
+ * <p>The owner referenced by this memory segment is null.
*
* @param buffer The byte buffer whose memory is represented by this memory segment.
* @throws IllegalArgumentException Thrown, if the given ByteBuffer is not direct.
@@ -61,13 +64,13 @@ public final class HybridMemorySegment extends MemorySegment {
HybridMemorySegment(ByteBuffer buffer) {
this(buffer, null);
}
-
+
/**
* Creates a new memory segment that represents the memory backing the given direct byte buffer.
* Note that the given ByteBuffer must be direct {@link java.nio.ByteBuffer#allocateDirect(int)},
* otherwise this method with throw an IllegalArgumentException.
- * <p>
- * The memory segment references the given owner.
+ *
+ * <p>The memory segment references the given owner.
*
* @param buffer The byte buffer whose memory is represented by this memory segment.
* @param owner The owner references by this memory segment.
@@ -80,8 +83,8 @@ public final class HybridMemorySegment extends MemorySegment {
/**
* Creates a new memory segment that represents the memory of the byte array.
- * <p>
- * The owner referenced by this memory segment is null.
+ *
+ * <p>The owner referenced by this memory segment is null.
*
* @param buffer The byte array whose memory is represented by this memory segment.
*/
@@ -91,8 +94,8 @@ public final class HybridMemorySegment extends MemorySegment {
/**
* Creates a new memory segment that represents the memory of the byte array.
- * <p>
- * The memory segment references the given owner.
+ *
+ * <p>The memory segment references the given owner.
*
* @param buffer The byte array whose memory is represented by this memory segment.
* @param owner The owner references by this memory segment.
@@ -105,7 +108,7 @@ public final class HybridMemorySegment extends MemorySegment {
// -------------------------------------------------------------------------
// MemorySegment operations
// -------------------------------------------------------------------------
-
+
public byte[] getArray() {
if (heapMemory != null) {
return heapMemory;
@@ -153,7 +156,7 @@ public final class HybridMemorySegment extends MemorySegment {
// ------------------------------------------------------------------------
// Random Access get() and put() methods
// ------------------------------------------------------------------------
-
+
@Override
public byte get(int index) {
final long pos = address + index;
@@ -197,7 +200,7 @@ public final class HybridMemorySegment extends MemorySegment {
@Override
public void get(int index, byte[] dst, int offset, int length) {
// check the byte array offset and length and the status
- if ( (offset | length | (offset + length) | (dst.length - (offset + length))) < 0) {
+ if ((offset | length | (offset + length) | (dst.length - (offset + length))) < 0) {
throw new IndexOutOfBoundsException();
}
@@ -221,7 +224,7 @@ public final class HybridMemorySegment extends MemorySegment {
if ((offset | length | (offset + length) | (src.length - (offset + length))) < 0) {
throw new IndexOutOfBoundsException();
}
-
+
final long pos = address + index;
if (index >= 0 && pos <= addressLimit - length) {
@@ -263,7 +266,7 @@ public final class HybridMemorySegment extends MemorySegment {
offset += 8;
length -= 8;
}
-
+
while (length > 0) {
out.writeByte(get(offset));
offset++;
@@ -396,7 +399,9 @@ public final class HybridMemorySegment extends MemorySegment {
// Utilities for native memory accesses and checks
// --------------------------------------------------------------------------------------------
- /** The reflection fields with which we access the off-heap pointer from direct ByteBuffers */
+ /**
+ * The reflection fields with which we access the off-heap pointer from direct ByteBuffers.
+ */
private static final Field ADDRESS_FIELD;
static {
@@ -421,7 +426,7 @@ public final class HybridMemorySegment extends MemorySegment {
throw new RuntimeException("Could not access direct byte buffer address.", t);
}
}
-
+
private static long checkBufferAndGetAddress(ByteBuffer buffer) {
if (buffer == null) {
throw new NullPointerException("buffer is null");
@@ -440,7 +445,7 @@ public final class HybridMemorySegment extends MemorySegment {
* Base factory for hybrid memory segments.
*/
public static final class HybridMemorySegmentFactory implements MemorySegmentFactory.Factory {
-
+
@Override
public HybridMemorySegment wrap(byte[] memory) {
return new HybridMemorySegment(memory);
@@ -461,9 +466,11 @@ public final class HybridMemorySegment extends MemorySegment {
return new HybridMemorySegment(memory, owner);
}
- /** prevent external instantiation */
+ /**
+ * Prevent external instantiation.
+ */
HybridMemorySegmentFactory() {}
- };
+ }
public static final HybridMemorySegmentFactory FACTORY = new HybridMemorySegmentFactory();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java
index d8315c5..1f3804f 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java
@@ -29,14 +29,14 @@ import java.nio.ByteOrder;
/**
* This class represents a piece of memory managed by Flink.
* The segment may be backed by heap memory (byte array) or by off-heap memory.
- * <p>
- * The methods for individual memory access are specialized in the classes
+ *
+ * <p>The methods for individual memory access are specialized in the classes
* {@link org.apache.flink.core.memory.HeapMemorySegment} and
* {@link org.apache.flink.core.memory.HybridMemorySegment}.
* All methods that operate across two memory segments are implemented in this class,
* to transparently handle the mixing of memory segment types.
- * <p>
- * This class fulfills conceptually a similar purpose as Java's {@link java.nio.ByteBuffer}.
+ *
+ * <p>This class fulfills conceptually a similar purpose as Java's {@link java.nio.ByteBuffer}.
* We add this specialized class for various reasons:
* <ul>
* <li>It offers additional binary compare, swap, and copy methods.</li>
@@ -47,20 +47,20 @@ import java.nio.ByteOrder;
* a byte order.</li>
* <li>It transparently and efficiently moves data between on-heap and off-heap variants.</li>
* </ul>
- *
- * <i>Comments on the implementation</i>:
+ *
+ * <p><i>Comments on the implementation</i>:
* We make heavy use of operations that are supported by native
* instructions, to achieve a high efficiency. Multi byte types (int, long, float, double, ...)
* are read and written with "unsafe" native commands.
- * <p>
- * Below is an example of the code generated for the {@link HeapMemorySegment#putLongBigEndian(int, long)}
+ *
+ * <p>Below is an example of the code generated for the {@link HeapMemorySegment#putLongBigEndian(int, long)}
* function by the just-in-time compiler. The code is grabbed from an Oracle JVM 7 using the
* hotspot disassembler library (hsdis32.dll) and the jvm command
* <i>-XX:+UnlockDiagnosticVMOptions -XX:CompileCommand=print,*MemorySegment.putLongBigEndian</i>.
* Note that this code realizes both the byte order swapping and the reinterpret cast access to
* get a long from the byte array.
*
- * <pre>
+ * <p><pre>
* [Verified Entry Point]
* 0x00007fc403e19920: sub $0x18,%rsp
* 0x00007fc403e19927: mov %rbp,0x10(%rsp) ;*synchronization entry
@@ -78,16 +78,16 @@ import java.nio.ByteOrder;
* 0x00007fc403e19943: pop %rbp
* 0x00007fc403e19944: test %eax,0x5ba76b6(%rip) # 0x00007fc4099c1000
* ; {poll_return}
- * 0x00007fc403e1994a: retq
+ * 0x00007fc403e1994a: retq
* </pre>
*
- * <i>Note on efficiency</i>:
+ * <p><i>Note on efficiency</i>:
* For best efficiency, the code that uses this class should make sure that only one
* subclass is loaded, or that the methods that are abstract in this class are used only from one of the
- * subclasses (either the {@link org.apache.flink.core.memory.HeapMemorySegment}, or the
+ * subclasses (either the {@link org.apache.flink.core.memory.HeapMemorySegment}, or the
* {@link org.apache.flink.core.memory.HybridMemorySegment}).
- *
- * That way, all the abstract methods in the MemorySegment base class have only one loaded
+ *
+ * <p>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 these conditions, the JIT can perfectly inline methods.
@@ -95,43 +95,62 @@ import java.nio.ByteOrder;
@Internal
public abstract class MemorySegment {
- /** The unsafe handle for transparent memory copied (heap / off-heap) */
+ /**
+ * The unsafe handle for transparent memory copied (heap / off-heap).
+ */
@SuppressWarnings("restriction")
protected static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
- /** The beginning of the byte array contents, relative to the byte array object */
+ /**
+ * The beginning of the byte array contents, relative to the byte array object.
+ */
@SuppressWarnings("restriction")
protected static final long BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
-
- /** Constant that flags the byte order. Because this is a boolean constant,
- * the JIT compiler can use this well to aggressively eliminate the non-applicable code paths */
+
+ /**
+ * Constant that flags the byte order. Because this is a boolean constant, the JIT compiler can
+ * use this well to aggressively eliminate the non-applicable code paths.
+ */
private static final boolean LITTLE_ENDIAN = (ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN);
-
+
// ------------------------------------------------------------------------
- /** The heap byte array object relative to which we access the memory. Is non-null if the
- * memory is on the heap, and is null, if the memory if off the heap. If we have this buffer, we
- * must never void this reference, or the memory segment will point to undefined addresses
- * outside the heap and may in out-of-order execution cases cause segmentation faults. */
+ /**
+ * The heap byte array object relative to which we access the memory.
+ *
+ * <p>Is non-<tt>null</tt> if the memory is on the heap, and is <tt>null</tt>, if the memory if
+ * off the heap. If we have this buffer, we must never void this reference, or the memory
+ * segment will point to undefined addresses outside the heap and may in out-of-order execution
+ * cases cause segmentation faults.
+ */
protected final byte[] heapMemory;
- /** The address to the data, relative to the heap memory byte array. If the heap memory byte array
- * is null, this becomes an absolute memory address outside the heap. */
+ /**
+ * The address to the data, relative to the heap memory byte array. If the heap memory byte
+ * array is <tt>null</tt>, this becomes an absolute memory address outside the heap.
+ */
protected long address;
- /** The address one byte after the last addressable byte.
- * This is address + size while the segment is not disposed */
+ /**
+ * The address one byte after the last addressable byte, i.e. <tt>address + size</tt> while the
+ * segment is not disposed.
+ */
protected final long addressLimit;
-
- /** The size in bytes of the memory segment */
+
+ /**
+ * The size in bytes of the memory segment.
+ */
protected final int size;
-
- /** Optional owner of the memory segment */
+
+ /**
+ * Optional owner of the memory segment.
+ */
private final Object owner;
/**
* Creates a new memory segment that represents the memory of the byte array.
- * Since the byte array is backed by on-heap memory, this memory segment holds its
+ *
+ * <p>Since the byte array is backed by on-heap memory, this memory segment holds its
* data on heap. The buffer must be at least of size 8 bytes.
*
* @param buffer The byte array whose memory is represented by this memory segment.
@@ -140,7 +159,7 @@ public abstract class MemorySegment {
if (buffer == null) {
throw new NullPointerException("buffer");
}
-
+
this.heapMemory = buffer;
this.address = BYTE_ARRAY_BASE_OFFSET;
this.size = buffer.length;
@@ -164,20 +183,21 @@ public abstract class MemorySegment {
throw new IllegalArgumentException("Segment initialized with too large address: " + offHeapAddress
+ " ; Max allowed address is " + (Long.MAX_VALUE - Integer.MAX_VALUE - 1));
}
-
+
this.heapMemory = null;
this.address = offHeapAddress;
this.addressLimit = this.address + size;
this.size = size;
this.owner = owner;
}
-
+
// ------------------------------------------------------------------------
// Memory Segment Operations
// ------------------------------------------------------------------------
/**
* Gets the size of the memory segment, in bytes.
+ *
* @return The size of the memory segment.
*/
public int size() {
@@ -186,16 +206,19 @@ public abstract class MemorySegment {
/**
* Checks whether the memory segment was freed.
- * @return True, if the memory segment has been freed, false otherwise.
+ *
+ * @return <tt>true</tt>, if the memory segment has been freed, <tt>false</tt> otherwise.
*/
public boolean isFreed() {
return address > addressLimit;
}
/**
- * Frees this memory segment. After this operation has been called, no further operations are
- * possible on the memory segment and will fail. The actual memory (heap or off-heap) will only
- * be released after this memory segment object has become garbage collected.
+ * Frees this memory segment.
+ *
+ * <p>After this operation has been called, no further operations are possible on the memory
+ * segment and will fail. The actual memory (heap or off-heap) will only be released after this
+ * memory segment object has become garbage collected.
*/
public void free() {
// this ensures we can place no more data and trigger
@@ -205,8 +228,9 @@ public abstract class MemorySegment {
/**
* Checks whether this memory segment is backed by off-heap memory.
- * @return True, if the memory segment is backed by off-heap memory, false if it is backed
- * by heap memory.
+ *
+ * @return <tt>true</tt>, if the memory segment is backed by off-heap memory, <tt>false</tt> if
+ * it is backed by heap memory.
*/
public boolean isOffHeap() {
return heapMemory == null;
@@ -218,6 +242,7 @@ public abstract class MemorySegment {
*
* @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.
@@ -226,13 +251,14 @@ public abstract class MemorySegment {
/**
* Gets the owner of this memory segment. Returns null, if the owner was not set.
+ *
* @return The owner of the memory segment, or null, if it does not have an owner.
*/
public Object getOwner() {
return owner;
}
-
-
+
+
// ------------------------------------------------------------------------
// Random Access get() and put() methods
// ------------------------------------------------------------------------
@@ -241,13 +267,13 @@ public abstract class MemorySegment {
// Notes on the implementation: We try to collapse as many checks as
// possible. We need to obey the following rules to make this safe
// against segfaults:
- //
+ //
// - Grab mutable fields onto the stack before checking and using. This
// guards us against concurrent modifications which invalidate the
// pointers
- // - Use subtrations for range checks, as they are tolerant
+ // - Use subtrations for range checks, as they are tolerant
//------------------------------------------------------------------------
-
+
/**
* Reads the byte at the given position.
*
@@ -277,7 +303,7 @@ public abstract class MemorySegment {
* @param index The position at which the first byte will be read.
* @param dst The memory into which the memory will be copied.
*
- * @throws IndexOutOfBoundsException Thrown, if the index is negative, or too large that the data between the
+ * @throws IndexOutOfBoundsException Thrown, if the index is negative, or too large that the data between the
* index and the memory segment end is not enough to fill the destination array.
*/
public abstract void get(int index, byte[] dst);
@@ -289,22 +315,22 @@ public abstract class MemorySegment {
* @param index The index in the memory segment array, where the data is put.
* @param src The source array to copy the data from.
*
- * @throws IndexOutOfBoundsException Thrown, if the index is negative, or too large such that the array
+ * @throws IndexOutOfBoundsException Thrown, if the index is negative, or too large such that the array
* size exceed the amount of memory between the index and the memory
- * segment's end.
+ * segment's end.
*/
public abstract void put(int index, byte[] src);
/**
* Bulk get method. Copies length memory from the specified position to the
- * destination memory, beginning at the given offset
+ * destination memory, beginning at the given offset.
*
* @param index The position at which the first byte will be read.
* @param dst The memory into which the memory will be copied.
* @param offset The copying offset in the destination memory.
* @param length The number of bytes to be copied.
*
- * @throws IndexOutOfBoundsException Thrown, if the index is negative, or too large that the requested number of
+ * @throws IndexOutOfBoundsException Thrown, if the index is negative, or too large that the requested number of
* bytes exceed the amount of memory between the index and the memory
* segment's end.
*/
@@ -320,7 +346,7 @@ public abstract class MemorySegment {
* @param offset The offset in the source array where the copying is started.
* @param length The number of bytes to copy.
*
- * @throws IndexOutOfBoundsException Thrown, if the index is negative, or too large such that the array
+ * @throws IndexOutOfBoundsException Thrown, if the index is negative, or too large such that the array
* portion to copy exceed the amount of memory between the index and the memory
* segment's end.
*/
@@ -377,7 +403,7 @@ public abstract class MemorySegment {
/**
* Reads an character value (16 bit, 2 bytes) from the given position, in little-endian byte order.
* This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getChar(int)}. For most cases (such as
+ * is possibly slower than {@link #getChar(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getChar(int)} is the preferable choice.
@@ -398,7 +424,7 @@ public abstract class MemorySegment {
/**
* Reads an character value (16 bit, 2 bytes) from the given position, in big-endian byte order.
* This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getChar(int)}. For most cases (such as
+ * is possibly slower than {@link #getChar(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getChar(int)} is the preferable choice.
@@ -443,7 +469,7 @@ public abstract class MemorySegment {
/**
* Writes the given character (16 bit, 2 bytes) to the given position in little-endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putChar(int, char)}. For most cases (such as
+ * is possibly slower than {@link #putChar(int, char)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putChar(int, char)} is the preferable choice.
@@ -464,7 +490,7 @@ public abstract class MemorySegment {
/**
* Writes the given character (16 bit, 2 bytes) to the given position in big-endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putChar(int, char)}. For most cases (such as
+ * is possibly slower than {@link #putChar(int, char)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putChar(int, char)} is the preferable choice.
@@ -509,7 +535,7 @@ public abstract class MemorySegment {
/**
* Reads an short integer value (16 bit, 2 bytes) from the given position, in little-endian byte order.
* This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getShort(int)}. For most cases (such as
+ * is possibly slower than {@link #getShort(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getShort(int)} is the preferable choice.
@@ -530,7 +556,7 @@ public abstract class MemorySegment {
/**
* Reads an short integer value (16 bit, 2 bytes) from the given position, in big-endian byte order.
* This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getShort(int)}. For most cases (such as
+ * is possibly slower than {@link #getShort(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getShort(int)} is the preferable choice.
@@ -575,7 +601,7 @@ public abstract class MemorySegment {
/**
* Writes the given short integer value (16 bit, 2 bytes) to the given position in little-endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putShort(int, short)}. For most cases (such as
+ * is possibly slower than {@link #putShort(int, short)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putShort(int, short)} is the preferable choice.
@@ -596,7 +622,7 @@ public abstract class MemorySegment {
/**
* Writes the given short integer value (16 bit, 2 bytes) to the given position in big-endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putShort(int, short)}. For most cases (such as
+ * is possibly slower than {@link #putShort(int, short)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putShort(int, short)} is the preferable choice.
@@ -618,7 +644,7 @@ public abstract class MemorySegment {
* Reads an int value (32bit, 4 bytes) from the given position, in the system's native byte order.
* This method offers the best speed for integer reading and should be used
* unless a specific byte order is required. In most cases, it suffices to know that the
- * byte order in which the value is written is the same as the one in which it is read
+ * byte order in which the value is written is the same as the one in which it is read
* (such as transient storage in memory, or serialization for I/O and network), making this
* method the preferable choice.
*
@@ -645,7 +671,7 @@ public abstract class MemorySegment {
/**
* Reads an int value (32bit, 4 bytes) from the given position, in little-endian byte order.
* This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getInt(int)}. For most cases (such as
+ * is possibly slower than {@link #getInt(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getInt(int)} is the preferable choice.
@@ -667,7 +693,7 @@ public abstract class MemorySegment {
/**
* Reads an int value (32bit, 4 bytes) from the given position, in big-endian byte order.
* This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getInt(int)}. For most cases (such as
+ * is possibly slower than {@link #getInt(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getInt(int)} is the preferable choice.
@@ -690,7 +716,7 @@ public abstract class MemorySegment {
* Writes the given int value (32bit, 4 bytes) to the given position in the system's native
* byte order. This method offers the best speed for integer writing and should be used
* unless a specific byte order is required. In most cases, it suffices to know that the
- * byte order in which the value is written is the same as the one in which it is read
+ * byte order in which the value is written is the same as the one in which it is read
* (such as transient storage in memory, or serialization for I/O and network), making this
* method the preferable choice.
*
@@ -717,7 +743,7 @@ public abstract class MemorySegment {
/**
* Writes the given int value (32bit, 4 bytes) to the given position in little endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putInt(int, int)}. For most cases (such as
+ * is possibly slower than {@link #putInt(int, int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putInt(int, int)} is the preferable choice.
@@ -739,7 +765,7 @@ public abstract class MemorySegment {
/**
* Writes the given int value (32bit, 4 bytes) to the given position in big endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putInt(int, int)}. For most cases (such as
+ * is possibly slower than {@link #putInt(int, int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putInt(int, int)} is the preferable choice.
@@ -762,7 +788,7 @@ public abstract class MemorySegment {
* Reads a long value (64bit, 8 bytes) from the given position, in the system's native byte order.
* This method offers the best speed for long integer reading and should be used
* unless a specific byte order is required. In most cases, it suffices to know that the
- * byte order in which the value is written is the same as the one in which it is read
+ * byte order in which the value is written is the same as the one in which it is read
* (such as transient storage in memory, or serialization for I/O and network), making this
* method the preferable choice.
*
@@ -789,7 +815,7 @@ public abstract class MemorySegment {
/**
* Reads a long integer value (64bit, 8 bytes) from the given position, in little endian byte order.
* This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getLong(int)}. For most cases (such as
+ * is possibly slower than {@link #getLong(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getLong(int)} is the preferable choice.
@@ -811,7 +837,7 @@ public abstract class MemorySegment {
/**
* Reads a long integer value (64bit, 8 bytes) from the given position, in big endian byte order.
* This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getLong(int)}. For most cases (such as
+ * is possibly slower than {@link #getLong(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getLong(int)} is the preferable choice.
@@ -834,7 +860,7 @@ public abstract class MemorySegment {
* Writes the given long value (64bit, 8 bytes) to the given position in the system's native
* byte order. This method offers the best speed for long integer writing and should be used
* unless a specific byte order is required. In most cases, it suffices to know that the
- * byte order in which the value is written is the same as the one in which it is read
+ * byte order in which the value is written is the same as the one in which it is read
* (such as transient storage in memory, or serialization for I/O and network), making this
* method the preferable choice.
*
@@ -861,7 +887,7 @@ public abstract class MemorySegment {
/**
* Writes the given long value (64bit, 8 bytes) to the given position in little endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putLong(int, long)}. For most cases (such as
+ * is possibly slower than {@link #putLong(int, long)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putLong(int, long)} is the preferable choice.
@@ -883,7 +909,7 @@ public abstract class MemorySegment {
/**
* Writes the given long value (64bit, 8 bytes) to the given position in big endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putLong(int, long)}. For most cases (such as
+ * is possibly slower than {@link #putLong(int, long)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putLong(int, long)} is the preferable choice.
@@ -906,7 +932,7 @@ public abstract class MemorySegment {
* Reads a single-precision floating point value (32bit, 4 bytes) from the given position, in the system's
* native byte order. This method offers the best speed for float reading and should be used
* unless a specific byte order is required. In most cases, it suffices to know that the
- * byte order in which the value is written is the same as the one in which it is read
+ * byte order in which the value is written is the same as the one in which it is read
* (such as transient storage in memory, or serialization for I/O and network), making this
* method the preferable choice.
*
@@ -923,7 +949,7 @@ public abstract class MemorySegment {
/**
* Reads a single-precision floating point value (32bit, 4 bytes) from the given position, in little endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getFloat(int)}. For most cases (such as
+ * is possibly slower than {@link #getFloat(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getFloat(int)} is the preferable choice.
@@ -941,7 +967,7 @@ public abstract class MemorySegment {
/**
* Reads a single-precision floating point value (32bit, 4 bytes) from the given position, in big endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getFloat(int)}. For most cases (such as
+ * is possibly slower than {@link #getFloat(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getFloat(int)} is the preferable choice.
@@ -960,7 +986,7 @@ public abstract class MemorySegment {
* Writes the given single-precision float value (32bit, 4 bytes) to the given position in the system's native
* byte order. This method offers the best speed for float writing and should be used
* unless a specific byte order is required. In most cases, it suffices to know that the
- * byte order in which the value is written is the same as the one in which it is read
+ * byte order in which the value is written is the same as the one in which it is read
* (such as transient storage in memory, or serialization for I/O and network), making this
* method the preferable choice.
*
@@ -977,7 +1003,7 @@ public abstract class MemorySegment {
/**
* Writes the given single-precision float value (32bit, 4 bytes) to the given position in little endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putFloat(int, float)}. For most cases (such as
+ * is possibly slower than {@link #putFloat(int, float)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putFloat(int, float)} is the preferable choice.
@@ -995,7 +1021,7 @@ public abstract class MemorySegment {
/**
* Writes the given single-precision float value (32bit, 4 bytes) to the given position in big endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putFloat(int, float)}. For most cases (such as
+ * is possibly slower than {@link #putFloat(int, float)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putFloat(int, float)} is the preferable choice.
@@ -1014,7 +1040,7 @@ public abstract class MemorySegment {
* Reads a double-precision floating point value (64bit, 8 bytes) from the given position, in the system's
* native byte order. This method offers the best speed for double reading and should be used
* unless a specific byte order is required. In most cases, it suffices to know that the
- * byte order in which the value is written is the same as the one in which it is read
+ * byte order in which the value is written is the same as the one in which it is read
* (such as transient storage in memory, or serialization for I/O and network), making this
* method the preferable choice.
*
@@ -1031,7 +1057,7 @@ public abstract class MemorySegment {
/**
* Reads a double-precision floating point value (64bit, 8 bytes) from the given position, in little endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getDouble(int)}. For most cases (such as
+ * is possibly slower than {@link #getDouble(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getDouble(int)} is the preferable choice.
@@ -1049,7 +1075,7 @@ public abstract class MemorySegment {
/**
* Reads a double-precision floating point value (64bit, 8 bytes) from the given position, in big endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #getDouble(int)}. For most cases (such as
+ * is possibly slower than {@link #getDouble(int)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #getDouble(int)} is the preferable choice.
@@ -1068,7 +1094,7 @@ public abstract class MemorySegment {
* Writes the given double-precision floating-point value (64bit, 8 bytes) to the given position in the
* system's native byte order. This method offers the best speed for double writing and should be used
* unless a specific byte order is required. In most cases, it suffices to know that the
- * byte order in which the value is written is the same as the one in which it is read
+ * byte order in which the value is written is the same as the one in which it is read
* (such as transient storage in memory, or serialization for I/O and network), making this
* method the preferable choice.
*
@@ -1085,7 +1111,7 @@ public abstract class MemorySegment {
/**
* Writes the given double-precision floating-point value (64bit, 8 bytes) to the given position in little endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putDouble(int, double)}. For most cases (such as
+ * is possibly slower than {@link #putDouble(int, double)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putDouble(int, double)} is the preferable choice.
@@ -1103,7 +1129,7 @@ public abstract class MemorySegment {
/**
* Writes the given double-precision floating-point value (64bit, 8 bytes) to the given position in big endian
* byte order. This method's speed depends on the system's native byte order, and it
- * is possibly slower than {@link #putDouble(int, double)}. For most cases (such as
+ * is possibly slower than {@link #putDouble(int, double)}. For most cases (such as
* transient storage in memory or serialization for I/O and network),
* it suffices to know that the byte order in which the value is written is the same as the
* one in which it is read, and {@link #putDouble(int, double)} is the preferable choice.
@@ -1130,7 +1156,7 @@ public abstract class MemorySegment {
*
* @param in The DataInput to get the data from.
* @param offset The position in the memory segment to copy the chunk to.
- * @param length The number of bytes to get.
+ * @param length The number of bytes to get.
*
* @throws IOException Thrown, if the DataInput encountered a problem upon reading,
* such as an End-Of-File.
@@ -1193,9 +1219,8 @@ public abstract class MemorySegment {
final long thisPointer = this.address + offset;
final long otherPointer = target.address + targetOffset;
- if ( (numBytes | offset | targetOffset) >= 0 &&
- thisPointer <= this.addressLimit - numBytes && otherPointer <= target.addressLimit - numBytes)
- {
+ if ((numBytes | offset | targetOffset) >= 0 &&
+ thisPointer <= this.addressLimit - numBytes && otherPointer <= target.addressLimit - numBytes) {
UNSAFE.copyMemory(thisHeapRef, thisPointer, otherHeapRef, otherPointer, numBytes);
}
else if (this.address > this.addressLimit) {
@@ -1262,17 +1287,17 @@ public abstract class MemorySegment {
* @param len Length of the swapped memory region
*/
public final void swapBytes(byte[] tempBuffer, MemorySegment seg2, int offset1, int offset2, int len) {
- if ( (offset1 | offset2 | len | (tempBuffer.length - len) ) >= 0) {
+ if ((offset1 | offset2 | len | (tempBuffer.length - len)) >= 0) {
final long thisPos = this.address + offset1;
final long otherPos = seg2.address + offset2;
-
+
if (thisPos <= this.addressLimit - len && otherPos <= seg2.addressLimit - len) {
// this -> temp buffer
UNSAFE.copyMemory(this.heapMemory, thisPos, tempBuffer, BYTE_ARRAY_BASE_OFFSET, len);
-
+
// other -> this
UNSAFE.copyMemory(seg2.heapMemory, otherPos, this.heapMemory, thisPos, len);
-
+
// temp buffer -> other
UNSAFE.copyMemory(tempBuffer, BYTE_ARRAY_BASE_OFFSET, seg2.heapMemory, otherPos, len);
return;
@@ -1284,7 +1309,7 @@ public abstract class MemorySegment {
throw new IllegalStateException("other memory segment has been freed.");
}
}
-
+
// index is in fact invalid
throw new IndexOutOfBoundsException(
String.format("offset1=%d, offset2=%d, len=%d, bufferSize=%d, address1=%d, address2=%d",
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
index 1e5c3ad..5411601 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
@@ -29,14 +29,16 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* for heap data are of the same type. That way, the runtime does not mix the various specializations
* of the {@link org.apache.flink.core.memory.MemorySegment}. Not mixing them has shown to be beneficial
* to method specialization by the JIT and to overall performance.
- *
+ *
* <p>Note that this factory auto-initialized to use {@link org.apache.flink.core.memory.HeapMemorySegment},
* if a request to create a segment comes before the initialization.
*/
@Internal
public class MemorySegmentFactory {
- /** The factory to use */
+ /**
+ * The factory to use.
+ */
private static volatile Factory factory;
/**
@@ -54,8 +56,8 @@ public class MemorySegmentFactory {
/**
* Allocates some unpooled memory and creates a new memory segment that represents
* that memory.
- * <p>
- * This method is similar to {@link #allocateUnpooledSegment(int, Object)}, but the
+ *
+ * <p>This method is similar to {@link #allocateUnpooledSegment(int, Object)}, but the
* memory segment will have null as the owner.
*
* @param size The size of the memory segment to allocate.
@@ -68,10 +70,10 @@ public class MemorySegmentFactory {
/**
* Allocates some unpooled memory and creates a new memory segment that represents
* that memory.
- * <p>
- * This method is similar to {@link #allocateUnpooledSegment(int)}, but additionally sets
+ *
+ * <p>This method is similar to {@link #allocateUnpooledSegment(int)}, but additionally sets
* the owner of the memory segment.
- *
+ *
* @param size The size of the memory segment to allocate.
* @param owner The owner to associate with the memory segment.
* @return A new memory segment, backed by unpooled heap memory.
@@ -83,11 +85,10 @@ public class MemorySegmentFactory {
/**
* Creates a memory segment that wraps the given byte array.
- * <p>
- * This method is intended to be used for components which pool memory and create
+ *
+ * <p>This method is intended to be used for components which pool memory and create
* memory segments around long-lived memory regions.
*
- *
* @param memory The heap memory to be represented by the memory segment.
* @param owner The owner to associate with the memory segment.
* @return A new memory segment representing the given heap memory.
@@ -99,9 +100,9 @@ public class MemorySegmentFactory {
/**
* Creates a memory segment that wraps the off-heap memory backing the given ByteBuffer.
- * Note that the ByteBuffer needs to be a <i>direct ByteBuffer</i>.
- * <p>
- * This method is intended to be used for components which pool memory and create
+ * Note that the ByteBuffer needs to be a <i>direct ByteBuffer</i>.
+ *
+ * <p>This method is intended to be used for components which pool memory and create
* memory segments around long-lived memory regions.
*
* @param memory The byte buffer with the off-heap memory to be represented by the memory segment.
@@ -140,7 +141,7 @@ public class MemorySegmentFactory {
/**
* Checks whether this memory segment factory has been initialized (with a type to produce).
- *
+ *
* @return True, if the factory has been initialized, false otherwise.
*/
public static boolean isInitialized() {
@@ -149,7 +150,7 @@ public class MemorySegmentFactory {
/**
* Gets the factory. May return null, if the factory has not been initialized.
- *
+ *
* @return The factory, or null, if the factory has not been initialized.
*/
public static Factory getFactory() {
@@ -157,7 +158,7 @@ public class MemorySegmentFactory {
}
/**
- * Sets the factory to the {@link HeapMemorySegment#FACTORY} is no factory has been initialized, yet.
+ * Sets the factory to the {@link HeapMemorySegment#FACTORY} if no factory has been initialized, yet.
*/
private static void ensureInitialized() {
if (factory == null) {
@@ -172,7 +173,7 @@ public class MemorySegmentFactory {
// ------------------------------------------------------------------------
// Internal factory
// ------------------------------------------------------------------------
-
+
/**
* A concrete factory for memory segments.
*/
@@ -198,8 +199,8 @@ public class MemorySegmentFactory {
/**
* Creates a memory segment that wraps the given byte array.
- * <p>
- * This method is intended to be used for components which pool memory and create
+ *
+ * <p>This method is intended to be used for components which pool memory and create
* memory segments around long-lived memory regions.
*
*
@@ -211,9 +212,9 @@ public class MemorySegmentFactory {
/**
* Creates a memory segment that wraps the off-heap memory backing the given ByteBuffer.
- * Note that the ByteBuffer needs to be a <i>direct ByteBuffer</i>.
- * <p>
- * This method is intended to be used for components which pool memory and create
+ * Note that the ByteBuffer needs to be a <i>direct ByteBuffer</i>.
+ *
+ * <p>This method is intended to be used for components which pool memory and create
* memory segments around long-lived memory regions.
*
* @param memory The byte buffer with the off-heap memory to be represented by the memory segment.
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentSource.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentSource.java b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentSource.java
index 8ae8299..2091b89 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentSource.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentSource.java
@@ -25,11 +25,11 @@ import org.apache.flink.annotation.Internal;
*/
@Internal
public interface MemorySegmentSource {
-
+
/**
* Gets the next memory segment. If no more segments are available, it returns null.
- *
+ *
* @return The next memory segment, or null, if none is available.
*/
- public MemorySegment nextSegment();
+ MemorySegment nextSegment();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java b/flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java
index b3a04cb..1cb507e 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java
@@ -28,11 +28,11 @@ import java.nio.ByteOrder;
*/
@Internal
public class MemoryUtils {
-
+
/** The "unsafe", which can be used to perform native memory accesses. */
@SuppressWarnings("restriction")
public static final sun.misc.Unsafe UNSAFE = getUnsafe();
-
+
/** The native byte order of the platform on which the system currently runs. */
public static final ByteOrder NATIVE_BYTE_ORDER = ByteOrder.nativeOrder();
@@ -55,6 +55,6 @@ public class MemoryUtils {
}
}
- /** Should not be instantiated */
+ /** Should not be instantiated. */
private MemoryUtils() {}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/SeekableDataInputView.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/SeekableDataInputView.java b/flink-core/src/main/java/org/apache/flink/core/memory/SeekableDataInputView.java
index 2c048de..9dab520 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/SeekableDataInputView.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/SeekableDataInputView.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.core.memory;
import org.apache.flink.annotation.PublicEvolving;
@@ -27,11 +26,11 @@ import org.apache.flink.annotation.PublicEvolving;
*/
@PublicEvolving
public interface SeekableDataInputView extends DataInputView {
-
+
/**
* Sets the read pointer to the given position.
- *
+ *
* @param position The new read position.
*/
- public void setReadPosition(long position);
+ void setReadPosition(long position);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/main/java/org/apache/flink/core/memory/SeekableDataOutputView.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/SeekableDataOutputView.java b/flink-core/src/main/java/org/apache/flink/core/memory/SeekableDataOutputView.java
index 967cefe..65b71db 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/SeekableDataOutputView.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/SeekableDataOutputView.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.core.memory;
import org.apache.flink.annotation.PublicEvolving;
@@ -27,11 +26,11 @@ import org.apache.flink.annotation.PublicEvolving;
*/
@PublicEvolving
public interface SeekableDataOutputView extends DataOutputView {
-
+
/**
* Sets the write pointer to the given position.
- *
+ *
* @param position The new write position.
*/
- public void setWritePosition(long position);
+ void setWritePosition(long position);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayInputStreamWithPosTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayInputStreamWithPosTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayInputStreamWithPosTest.java
index 1e1902e..4797c0a 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayInputStreamWithPosTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayInputStreamWithPosTest.java
@@ -24,12 +24,15 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
+/**
+ * Tests for {@link ByteArrayInputStreamWithPos}.
+ */
public class ByteArrayInputStreamWithPosTest {
@Rule
public ExpectedException thrown = ExpectedException.none();
- private byte[] data = new byte[] {'0','1','2','3','4','5','6','7','8','9'};
+ private byte[] data = new byte[] {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9'};
private ByteArrayInputStreamWithPos stream;
@@ -39,7 +42,7 @@ public class ByteArrayInputStreamWithPosTest {
}
/**
- * Test setting position on a {@link ByteArrayInputStreamWithPos}
+ * Test setting position on a {@link ByteArrayInputStreamWithPos}.
*/
@Test
public void testSetPosition() throws Exception {
@@ -60,7 +63,7 @@ public class ByteArrayInputStreamWithPosTest {
}
/**
- * Test that the expected position exceeds the capacity of the byte array
+ * Test that the expected position exceeds the capacity of the byte array.
*/
@Test
public void testSetTooLargePosition() throws Exception {
@@ -70,7 +73,7 @@ public class ByteArrayInputStreamWithPosTest {
}
/**
- * Test setting a negative position
+ * Test setting a negative position.
*/
@Test
public void testSetNegativePosition() throws Exception {
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayOutputStreamWithPosTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayOutputStreamWithPosTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayOutputStreamWithPosTest.java
index 3c2e361..b076d08 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayOutputStreamWithPosTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/ByteArrayOutputStreamWithPosTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.core.memory;
import org.apache.flink.configuration.ConfigConstants;
+
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
@@ -28,6 +29,9 @@ import org.junit.rules.ExpectedException;
import java.io.IOException;
import java.util.Arrays;
+/**
+ * Tests for {@link ByteArrayOutputStreamWithPos}.
+ */
public class ByteArrayOutputStreamWithPosTest {
private static final int BUFFER_SIZE = 32;
@@ -43,7 +47,7 @@ public class ByteArrayOutputStreamWithPosTest {
}
/**
- * Test setting position which is exactly the same with the buffer size
+ * Test setting position which is exactly the same with the buffer size.
*/
@Test
public void testSetPositionWhenBufferIsFull() throws Exception {
@@ -62,7 +66,7 @@ public class ByteArrayOutputStreamWithPosTest {
}
/**
- * Test setting negative position
+ * Test setting negative position.
*/
@Test
public void testSetNegativePosition() throws Exception {
@@ -74,7 +78,7 @@ public class ByteArrayOutputStreamWithPosTest {
}
/**
- * Test setting position larger than buffer size
+ * Test setting position larger than buffer size.
*/
@Test
public void testSetPositionLargerThanBufferSize() throws Exception {
@@ -89,7 +93,7 @@ public class ByteArrayOutputStreamWithPosTest {
}
/**
- * Test that toString returns a substring of the buffer with range(0, position)
+ * Test that toString returns a substring of the buffer with range(0, position).
*/
@Test
public void testToString() throws IOException {
@@ -100,7 +104,7 @@ public class ByteArrayOutputStreamWithPosTest {
stream.write(data);
Assert.assertArrayEquals(data, stream.toString().getBytes(ConfigConstants.DEFAULT_CHARSET));
- for (int i = 0 ; i < data.length ; i++) {
+ for (int i = 0; i < data.length; i++) {
stream.setPosition(i);
Assert.assertArrayEquals(Arrays.copyOf(data, i), stream.toString().getBytes(ConfigConstants.DEFAULT_CHARSET));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/CrossSegmentTypeTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/CrossSegmentTypeTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/CrossSegmentTypeTest.java
index 0e3fed5..ea144c7 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/CrossSegmentTypeTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/CrossSegmentTypeTest.java
@@ -29,12 +29,16 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+/**
+ * Verifies interoperability between {@link HeapMemorySegment} and {@link HybridMemorySegment} (in
+ * both heap and off-heap modes).
+ */
public class CrossSegmentTypeTest {
- private final int pageSize = 32*1024;
-
+ private final int pageSize = 32 * 1024;
+
// ------------------------------------------------------------------------
-
+
@Test
public void testCompareBytesMixedSegments() {
MemorySegment[] segs1 = {
@@ -57,11 +61,11 @@ public class CrossSegmentTypeTest {
}
}
}
-
+
private void testCompare(MemorySegment seg1, MemorySegment seg2, Random random) {
assertEquals(pageSize, seg1.size());
assertEquals(pageSize, seg2.size());
-
+
final byte[] bytes1 = new byte[pageSize];
final byte[] bytes2 = new byte[pageSize];
@@ -76,7 +80,7 @@ public class CrossSegmentTypeTest {
bytes2[i + shift] = val;
}
}
-
+
seg1.put(0, bytes1);
seg2.put(0, bytes2);
@@ -85,7 +89,7 @@ public class CrossSegmentTypeTest {
int pos2 = random.nextInt(bytes2.length);
int len = Math.min(Math.min(bytes1.length - pos1, bytes2.length - pos2),
- random.nextInt(pageSize / 50 ));
+ random.nextInt(pageSize / 50));
int cmp = seg1.compare(seg2, pos1, pos2, len);
@@ -98,10 +102,9 @@ public class CrossSegmentTypeTest {
}
}
-
@Test
public void testSwapBytesMixedSegments() {
- final int HALF_SIZE = pageSize / 2;
+ final int halfPageSize = pageSize / 2;
MemorySegment[] segs1 = {
new HeapMemorySegment(new byte[pageSize]),
@@ -110,20 +113,20 @@ public class CrossSegmentTypeTest {
};
MemorySegment[] segs2 = {
- new HeapMemorySegment(new byte[HALF_SIZE]),
- new HybridMemorySegment(new byte[HALF_SIZE]),
- new HybridMemorySegment(ByteBuffer.allocateDirect(HALF_SIZE))
+ new HeapMemorySegment(new byte[halfPageSize]),
+ new HybridMemorySegment(new byte[halfPageSize]),
+ new HybridMemorySegment(ByteBuffer.allocateDirect(halfPageSize))
};
Random rnd = new Random();
for (MemorySegment seg1 : segs1) {
for (MemorySegment seg2 : segs2) {
- testSwap(seg1, seg2, rnd, HALF_SIZE);
+ testSwap(seg1, seg2, rnd, halfPageSize);
}
}
}
-
+
private void testSwap(MemorySegment seg1, MemorySegment seg2, Random random, int smallerSize) {
assertEquals(pageSize, seg1.size());
assertEquals(smallerSize, seg2.size());
@@ -132,7 +135,7 @@ public class CrossSegmentTypeTest {
final byte[] bytes2 = new byte[smallerSize];
Arrays.fill(bytes2, (byte) 1);
-
+
seg1.put(0, bytes1);
seg2.put(0, bytes2);
@@ -184,29 +187,29 @@ public class CrossSegmentTypeTest {
byte[] expected = new byte[pageSize];
byte[] actual = new byte[pageSize];
-
+
// zero out the memory
seg1.put(0, expected);
seg2.put(0, expected);
-
+
for (int i = 0; i < 40; i++) {
int numBytes = random.nextInt(pageSize / 20);
byte[] bytes = new byte[numBytes];
random.nextBytes(bytes);
-
+
int thisPos = random.nextInt(pageSize - numBytes);
int otherPos = random.nextInt(pageSize - numBytes);
-
+
// track what we expect
System.arraycopy(bytes, 0, expected, otherPos, numBytes);
-
+
seg1.put(thisPos, bytes);
seg1.copyTo(thisPos, seg2, otherPos, numBytes);
}
-
+
seg2.get(0, actual);
assertArrayEquals(expected, actual);
-
+
// test out of bound conditions
final int[] validOffsets = { 0, 1, pageSize / 10 * 9 };
@@ -229,7 +232,7 @@ public class CrossSegmentTypeTest {
fail("should fail with an IndexOutOfBoundsException");
}
catch (IndexOutOfBoundsException ignored) {}
-
+
try {
seg2.copyTo(off1, seg1, off2, len);
fail("should fail with an IndexOutOfBoundsException");
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/EndiannessAccessChecks.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/EndiannessAccessChecks.java b/flink-core/src/test/java/org/apache/flink/core/memory/EndiannessAccessChecks.java
index c06562a..c04d435 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/EndiannessAccessChecks.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/EndiannessAccessChecks.java
@@ -18,14 +18,19 @@
package org.apache.flink.core.memory;
+import org.junit.Test;
+
import java.nio.ByteBuffer;
import java.util.Random;
-import org.junit.Test;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+/**
+ * Verifies correct accesses with regards to endianness in {@link HeapMemorySegment} and {@link
+ * HybridMemorySegment} (in both heap and off-heap modes).
+ */
public class EndiannessAccessChecks {
-
+
@Test
public void testHeapSegment() {
testBigAndLittleEndianAccessUnaligned(new HeapMemorySegment(new byte[11111]));
@@ -40,19 +45,19 @@ public class EndiannessAccessChecks {
public void testHybridOffHeapSegment() {
testBigAndLittleEndianAccessUnaligned(new HybridMemorySegment(ByteBuffer.allocateDirect(11111)));
}
-
+
private void testBigAndLittleEndianAccessUnaligned(MemorySegment segment) {
final Random rnd = new Random();
-
+
// longs
{
final long seed = rnd.nextLong();
-
+
rnd.setSeed(seed);
for (int i = 0; i < 10000; i++) {
long val = rnd.nextLong();
int pos = rnd.nextInt(segment.size - 7);
-
+
segment.putLongLittleEndian(pos, val);
long r = segment.getLongBigEndian(pos);
assertEquals(val, Long.reverseBytes(r));
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/HeapMemorySegmentTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/HeapMemorySegmentTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/HeapMemorySegmentTest.java
index b2c16d9..90176fd 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/HeapMemorySegmentTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/HeapMemorySegmentTest.java
@@ -24,8 +24,13 @@ import org.junit.runners.Parameterized;
import java.nio.ByteBuffer;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+/**
+ * Tests for the {@link HeapMemorySegment} in off-heap mode.
+ */
@RunWith(Parameterized.class)
public class HeapMemorySegmentTest extends MemorySegmentTestBase {
@@ -42,7 +47,7 @@ public class HeapMemorySegmentTest extends MemorySegmentTestBase {
MemorySegment createSegment(int size, Object owner) {
return new HeapMemorySegment(new byte[size], owner);
}
-
+
@Test
public void testHeapSegmentSpecifics() {
final byte[] buffer = new byte[411];
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/HybridOffHeapMemorySegmentTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/HybridOffHeapMemorySegmentTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/HybridOffHeapMemorySegmentTest.java
index cf2d023..254254c 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/HybridOffHeapMemorySegmentTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/HybridOffHeapMemorySegmentTest.java
@@ -29,6 +29,9 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+/**
+ * Tests for the {@link HybridMemorySegment} in off-heap mode.
+ */
@RunWith(Parameterized.class)
public class HybridOffHeapMemorySegmentTest extends MemorySegmentTestBase {
@@ -57,6 +60,7 @@ public class HybridOffHeapMemorySegmentTest extends MemorySegmentTestBase {
assertTrue(buffer == seg.getOffHeapBuffer());
try {
+ //noinspection ResultOfMethodCallIgnored
seg.getArray();
fail("should throw an exception");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/HybridOnHeapMemorySegmentTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/HybridOnHeapMemorySegmentTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/HybridOnHeapMemorySegmentTest.java
index 711326f..6672744 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/HybridOnHeapMemorySegmentTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/HybridOnHeapMemorySegmentTest.java
@@ -29,6 +29,9 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+/**
+ * Tests for the {@link HybridMemorySegment} in on-heap mode.
+ */
@RunWith(Parameterized.class)
public class HybridOnHeapMemorySegmentTest extends MemorySegmentTestBase {
@@ -45,7 +48,7 @@ public class HybridOnHeapMemorySegmentTest extends MemorySegmentTestBase {
MemorySegment createSegment(int size, Object owner) {
return new HybridMemorySegment(new byte[size], owner);
}
-
+
@Test
public void testHybridHeapSegmentSpecifics() {
final byte[] buffer = new byte[411];
@@ -57,6 +60,7 @@ public class HybridOnHeapMemorySegmentTest extends MemorySegmentTestBase {
assertTrue(buffer == seg.getArray());
try {
+ //noinspection ResultOfMethodCallIgnored
seg.getOffHeapBuffer();
fail("should throw an exception");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentChecksTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentChecksTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentChecksTest.java
index f50322c..ccf3a4b 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentChecksTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentChecksTest.java
@@ -28,7 +28,7 @@ import java.nio.ByteBuffer;
* Tests for the sanity checks of the memory segments.
*/
public class MemorySegmentChecksTest {
-
+
@Test(expected = NullPointerException.class)
public void testHeapNullBuffer1() {
new HeapMemorySegment(null);
@@ -66,21 +66,21 @@ public class MemorySegmentChecksTest {
@Test(expected = IllegalArgumentException.class)
public void testZeroAddress(){
- new MockSegment(0L, 4*1024, null);
+ new MockSegment(0L, 4 * 1024, null);
}
@Test(expected = IllegalArgumentException.class)
public void testNegativeAddress(){
- new MockSegment(-1L, 4*1024, null);
+ new MockSegment(-1L, 4 * 1024, null);
}
@Test(expected = IllegalArgumentException.class)
public void testTooLargeAddress(){
- new MockSegment(Long.MAX_VALUE - 8*1024, 4*1024, null);
+ new MockSegment(Long.MAX_VALUE - 8 * 1024, 4 * 1024, null);
}
-
+
// ------------------------------------------------------------------------
-
+
final class MockSegment extends MemorySegment {
MockSegment(long offHeapAddress, int size, Object owner) {
@@ -131,5 +131,5 @@ public class MemorySegmentChecksTest {
@Override
public void put(int offset, ByteBuffer source, int numBytes) {}
- };
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java
index 84105cf..3fd2cbe 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java
@@ -22,8 +22,13 @@ import org.junit.Test;
import java.lang.reflect.Field;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+/**
+ * Tests for the {@link MemorySegmentFactory} in on/off-heap modes.
+ */
public class MemorySegmentFactoryTest {
@Test
[06/10] flink git commit: [FLINK-7739] [kafka connector] Fix test
instabilities
Posted by se...@apache.org.
[FLINK-7739] [kafka connector] Fix test instabilities
- Set shorter heartbeats intervals. Default pause value of 60seconds is
too large (tests would timeout before akka react)
- Exclude netty dependency from zookeeper. Zookeeper was pulling in
conflicting Netty version. Conflict was extremly subtle - TaskManager in
Kafka tests was deadlocking in some rare corner cases.
This closes #4775
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3581a335
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3581a335
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3581a335
Branch: refs/heads/master
Commit: 3581a3350892a5c584a617d1b66cffdada4e17dd
Parents: a8443a5
Author: Piotr Nowojski <pi...@gmail.com>
Authored: Fri Sep 29 18:23:29 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 20:12:51 2017 +0200
----------------------------------------------------------------------
.../apache/flink/streaming/connectors/kafka/KafkaTestBase.java | 3 +++
pom.xml | 6 ++++++
2 files changed, 9 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3581a335/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
index 19f38e2..f8792e5 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MetricOptions;
@@ -121,6 +122,8 @@ public abstract class KafkaTestBase extends TestLogger {
protected static Configuration getFlinkConfiguration() {
Configuration flinkConfig = new Configuration();
+ flinkConfig.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, "5 s");
+ flinkConfig.setString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL, "1 s");
flinkConfig.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TMS);
flinkConfig.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, TM_SLOTS);
flinkConfig.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 16L);
http://git-wip-us.apache.org/repos/asf/flink/blob/3581a335/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 7f09abc..524e718 100644
--- a/pom.xml
+++ b/pom.xml
@@ -460,6 +460,12 @@ under the License.
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
+ <!-- Netty from zookeeper is conflicting with akka's, which can cause occasional deadlock inside
+ Netty threads -->
+ <exclusion>
+ <groupId>io.netty</groupId>
+ <artifactId>netty</artifactId>
+ </exclusion>
</exclusions>
</dependency>
[03/10] flink git commit: [hotfix] [build] Remove no longer needed
explicit Hadoop exclusions
Posted by se...@apache.org.
[hotfix] [build] Remove no longer needed explicit Hadoop exclusions
Now that Hadoop is an optinal dependency, all the explicit exclusions
during shading are no longer needed.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d3e15fd9
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d3e15fd9
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d3e15fd9
Branch: refs/heads/master
Commit: d3e15fd976f4cb7da89928dbea1b81e4d28a918e
Parents: 8e02df0
Author: Stephan Ewen <se...@apache.org>
Authored: Fri Oct 6 14:25:23 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 20:12:51 2017 +0200
----------------------------------------------------------------------
flink-dist/pom.xml | 60 -------------------------------------------------
1 file changed, 60 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d3e15fd9/flink-dist/pom.xml
----------------------------------------------------------------------
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index b874c49..aca4c46 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -47,72 +47,36 @@ under the License.
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-hadoop2</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_${scala.binary.version}</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-hadoop2</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime_${scala.binary.version}</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-hadoop2</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-hadoop2</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-optimizer_${scala.binary.version}</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-hadoop2</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-hadoop2</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
@@ -125,48 +89,24 @@ under the License.
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-hadoop2</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-hadoop2</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-python_${scala.binary.version}</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-hadoop2</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala-shell_${scala.binary.version}</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-hadoop2</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
[04/10] flink git commit: [hotfix] [build] Properly exclude
flink-python artifact from fat dist jar
Posted by se...@apache.org.
[hotfix] [build] Properly exclude flink-python artifact from fat dist jar
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/774e7f79
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/774e7f79
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/774e7f79
Branch: refs/heads/master
Commit: 774e7f797bf1601a0e9c5971f71336020895fba6
Parents: 77e3701
Author: Stephan Ewen <se...@apache.org>
Authored: Fri Oct 6 13:22:19 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 20:12:51 2017 +0200
----------------------------------------------------------------------
flink-dist/pom.xml | 2 +-
tools/travis_mvn_watchdog.sh | 11 +++++++++--
2 files changed, 10 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/774e7f79/flink-dist/pom.xml
----------------------------------------------------------------------
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index 8120cd8..0490dcc 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -506,7 +506,7 @@ under the License.
<exclude>org.apache.flink:flink-examples-batch</exclude>
<exclude>org.apache.flink:flink-examples-streaming</exclude>
<exclude>org.apache.flink:flink-examples-table</exclude>
- <exclude>org.apache.flink:flink-python</exclude>
+ <exclude>org.apache.flink:flink-python_${scala.binary.version}</exclude>
<exclude>org.slf4j:slf4j-log4j12</exclude>
<exclude>log4j:log4j</exclude>
</excludes>
http://git-wip-us.apache.org/repos/asf/flink/blob/774e7f79/tools/travis_mvn_watchdog.sh
----------------------------------------------------------------------
diff --git a/tools/travis_mvn_watchdog.sh b/tools/travis_mvn_watchdog.sh
index dd75d36..3ecc268 100755
--- a/tools/travis_mvn_watchdog.sh
+++ b/tools/travis_mvn_watchdog.sh
@@ -301,11 +301,18 @@ check_shaded_artifacts() {
return 1
fi
+ FLINK_PYTHON=`cat allClasses | grep '^org/apache/flink/python' | wc -l`
+ if [ "$FLINK_PYTHON" != "0" ]; then
+ echo "=============================================================================="
+ echo "Detected that the Flink Python artifact is in the dist jar"
+ echo "=============================================================================="
+ return 1
+ fi
HADOOP=`cat allClasses | grep '^org/apache/hadoop' | wc -l`
if [ "$HADOOP" != "0" ]; then
echo "=============================================================================="
- echo "Detected '$HADOOP' Hadoop classes in the dist jar
+ echo "Detected '$HADOOP' Hadoop classes in the dist jar"
echo "=============================================================================="
return 1
fi
@@ -313,7 +320,7 @@ check_shaded_artifacts() {
MAPR=`cat allClasses | grep '^com/mapr' | wc -l`
if [ "$MAPR" != "0" ]; then
echo "=============================================================================="
- echo "Detected '$MAPR' MapR classes in the dist jar
+ echo "Detected '$MAPR' MapR classes in the dist jar"
echo "=============================================================================="
return 1
fi
[05/10] flink git commit: [FLINK-7742] [java api] [runtime] Fix array
index out of bounds exceptions
Posted by se...@apache.org.
[FLINK-7742] [java api] [runtime] Fix array index out of bounds exceptions
This closes #4754
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/edc53a74
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/edc53a74
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/edc53a74
Branch: refs/heads/master
Commit: edc53a74553f19c2ec83ab7c12945758ba6084bf
Parents: 3581a33
Author: yew1eb <ye...@gmail.com>
Authored: Sat Sep 30 14:50:59 2017 +0800
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 20:12:51 2017 +0200
----------------------------------------------------------------------
flink-java/src/main/java/org/apache/flink/api/java/Utils.java | 2 +-
.../runtime/query/netty/message/KvStateRequestSerializer.java | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/edc53a74/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
index 44e176c..0184e58 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
@@ -53,7 +53,7 @@ public final class Utils {
public static String getCallLocationName(int depth) {
StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
- if (stackTrace.length < depth) {
+ if (stackTrace.length <= depth) {
return "<unknown>";
}
http://git-wip-us.apache.org/repos/asf/flink/blob/edc53a74/flink-runtime/src/main/java/org/apache/flink/runtime/query/netty/message/KvStateRequestSerializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/query/netty/message/KvStateRequestSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/query/netty/message/KvStateRequestSerializer.java
index f0cc94a..68f06e3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/query/netty/message/KvStateRequestSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/query/netty/message/KvStateRequestSerializer.java
@@ -224,7 +224,7 @@ public final class KvStateRequestSerializer {
// Get the message type
int msgType = buf.readInt();
KvStateRequestType[] values = KvStateRequestType.values();
- if (msgType >= 0 && msgType <= values.length) {
+ if (msgType >= 0 && msgType < values.length) {
return values[msgType];
} else {
throw new IllegalArgumentException("Illegal message type with index " + msgType);
[02/10] flink git commit: [FLINK-7772][blob] fix test instability in
BlobCacheDeleteTest
Posted by se...@apache.org.
[FLINK-7772][blob] fix test instability in BlobCacheDeleteTest
BlobCacheDeleteTest did not account for the server executing the delete call of
a transient BLOB after acknowledging the request. This resulted in the
testDeleteTransientLocalFails* tests failing.
This closes #4782
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a8443a52
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a8443a52
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a8443a52
Branch: refs/heads/master
Commit: a8443a52e33e5cf1f85b3be7689292cf52465e40
Parents: d3e15fd
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Fri Oct 6 12:34:40 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 20:12:51 2017 +0200
----------------------------------------------------------------------
.../apache/flink/runtime/blob/BlobCacheDeleteTest.java | 13 ++++++++-----
1 file changed, 8 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a8443a52/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
index 8ea2a5e..a83d100 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
@@ -45,6 +45,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import static org.apache.flink.runtime.blob.BlobCachePutTest.verifyDeletedEventually;
import static org.apache.flink.runtime.blob.BlobKey.BlobType.TRANSIENT_BLOB;
import static org.apache.flink.runtime.blob.BlobServerDeleteTest.delete;
import static org.apache.flink.runtime.blob.BlobServerGetTest.verifyDeleted;
@@ -208,12 +209,12 @@ public class BlobCacheDeleteTest extends TestLogger {
}
@Test
- public void testDeleteTransientLocalFailsNoJob() throws IOException {
+ public void testDeleteTransientLocalFailsNoJob() throws IOException, InterruptedException {
testDeleteTransientLocalFails(null);
}
@Test
- public void testDeleteTransientLocalFailsForJob() throws IOException {
+ public void testDeleteTransientLocalFailsForJob() throws IOException, InterruptedException {
testDeleteTransientLocalFails(new JobID());
}
@@ -225,7 +226,8 @@ public class BlobCacheDeleteTest extends TestLogger {
* @param jobId
* job id
*/
- private void testDeleteTransientLocalFails(@Nullable final JobID jobId) throws IOException {
+ private void testDeleteTransientLocalFails(@Nullable final JobID jobId)
+ throws IOException, InterruptedException {
assumeTrue(!OperatingSystem.isWindows()); //setWritable doesn't work on Windows.
final Configuration config = new Configuration();
@@ -259,10 +261,11 @@ public class BlobCacheDeleteTest extends TestLogger {
// issue a DELETE request
assertFalse(delete(cache, jobId, key));
- verifyDeleted(server, jobId, key);
- // the file should still be there on both cache and server
+ // the file should still be there on the cache
verifyContents(cache, jobId, key, data);
+ // the server should have started the delete call after the cache accessed the (transient!) BLOB
+ verifyDeletedEventually(server, jobId, key);
} finally {
if (blobFile != null && directory != null) {
//noinspection ResultOfMethodCallIgnored
[08/10] flink git commit: [FLINK-7312] [checkstyle] Enable checkstyle
for 'flink/core/memory/*'
Posted by se...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentTestBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentTestBase.java b/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentTestBase.java
index 8c10902..62e0a56 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentTestBase.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentTestBase.java
@@ -36,19 +36,24 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Random;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
/**
* Tests for the access and transfer methods of the HeapMemorySegment.
*/
public abstract class MemorySegmentTestBase {
-
+
private final Random random = new Random();
-
+
private final int pageSize;
-
-
- public MemorySegmentTestBase(int pageSize) {
+
+ MemorySegmentTestBase(int pageSize) {
this.pageSize = pageSize;
}
@@ -58,8 +63,8 @@ public abstract class MemorySegmentTestBase {
abstract MemorySegment createSegment(int size);
- abstract MemorySegment createSegment(int size, Object owner);
-
+ abstract MemorySegment createSegment(@SuppressWarnings("SameParameterValue") int size, Object owner);
+
// ------------------------------------------------------------------------
// Access to primitives
// ------------------------------------------------------------------------
@@ -409,7 +414,7 @@ public abstract class MemorySegmentTestBase {
continue;
} else {
occupied[pos] = true;
- occupied[pos+1] = true;
+ occupied[pos + 1] = true;
}
segment.putChar(pos, (char) (random.nextInt(Character.MAX_VALUE)));
@@ -425,7 +430,7 @@ public abstract class MemorySegmentTestBase {
continue;
} else {
occupied[pos] = true;
- occupied[pos+1] = true;
+ occupied[pos + 1] = true;
}
assertEquals((char) (random.nextInt(Character.MAX_VALUE)), segment.getChar(pos));
@@ -544,7 +549,7 @@ public abstract class MemorySegmentTestBase {
continue;
} else {
occupied[pos] = true;
- occupied[pos+1] = true;
+ occupied[pos + 1] = true;
}
segment.putShort(pos, (short) random.nextInt());
@@ -560,7 +565,7 @@ public abstract class MemorySegmentTestBase {
continue;
} else {
occupied[pos] = true;
- occupied[pos+1] = true;
+ occupied[pos + 1] = true;
}
assertEquals((short) random.nextInt(), segment.getShort(pos));
@@ -695,9 +700,9 @@ public abstract class MemorySegmentTestBase {
continue;
} else {
occupied[pos] = true;
- occupied[pos+1] = true;
- occupied[pos+2] = true;
- occupied[pos+3] = true;
+ occupied[pos + 1] = true;
+ occupied[pos + 2] = true;
+ occupied[pos + 3] = true;
}
segment.putInt(pos, random.nextInt());
@@ -713,9 +718,9 @@ public abstract class MemorySegmentTestBase {
continue;
} else {
occupied[pos] = true;
- occupied[pos+1] = true;
- occupied[pos+2] = true;
- occupied[pos+3] = true;
+ occupied[pos + 1] = true;
+ occupied[pos + 2] = true;
+ occupied[pos + 3] = true;
}
assertEquals(random.nextInt(), segment.getInt(pos));
@@ -847,19 +852,18 @@ public abstract class MemorySegmentTestBase {
int pos = random.nextInt(pageSize - 7);
if (occupied[pos] || occupied[pos + 1] || occupied[pos + 2] || occupied[pos + 3] ||
- occupied[pos + 4] || occupied[pos + 5] || occupied[pos + 6] || occupied[pos + 7])
- {
+ occupied[pos + 4] || occupied[pos + 5] || occupied[pos + 6] || occupied[pos + 7]) {
continue;
}
else {
occupied[pos] = true;
- occupied[pos+1] = true;
- occupied[pos+2] = true;
- occupied[pos+3] = true;
- occupied[pos+4] = true;
- occupied[pos+5] = true;
- occupied[pos+6] = true;
- occupied[pos+7] = true;
+ occupied[pos + 1] = true;
+ occupied[pos + 2] = true;
+ occupied[pos + 3] = true;
+ occupied[pos + 4] = true;
+ occupied[pos + 5] = true;
+ occupied[pos + 6] = true;
+ occupied[pos + 7] = true;
}
segment.putLong(pos, random.nextLong());
@@ -872,19 +876,18 @@ public abstract class MemorySegmentTestBase {
int pos = random.nextInt(pageSize - 7);
if (occupied[pos] || occupied[pos + 1] || occupied[pos + 2] || occupied[pos + 3] ||
- occupied[pos + 4] || occupied[pos + 5] || occupied[pos + 6] || occupied[pos + 7])
- {
+ occupied[pos + 4] || occupied[pos + 5] || occupied[pos + 6] || occupied[pos + 7]) {
continue;
}
else {
occupied[pos] = true;
- occupied[pos+1] = true;
- occupied[pos+2] = true;
- occupied[pos+3] = true;
- occupied[pos+4] = true;
- occupied[pos+5] = true;
- occupied[pos+6] = true;
- occupied[pos+7] = true;
+ occupied[pos + 1] = true;
+ occupied[pos + 2] = true;
+ occupied[pos + 3] = true;
+ occupied[pos + 4] = true;
+ occupied[pos + 5] = true;
+ occupied[pos + 6] = true;
+ occupied[pos + 7] = true;
}
assertEquals(random.nextLong(), segment.getLong(pos));
@@ -1019,9 +1022,9 @@ public abstract class MemorySegmentTestBase {
continue;
} else {
occupied[pos] = true;
- occupied[pos+1] = true;
- occupied[pos+2] = true;
- occupied[pos+3] = true;
+ occupied[pos + 1] = true;
+ occupied[pos + 2] = true;
+ occupied[pos + 3] = true;
}
segment.putFloat(pos, random.nextFloat());
@@ -1037,9 +1040,9 @@ public abstract class MemorySegmentTestBase {
continue;
} else {
occupied[pos] = true;
- occupied[pos+1] = true;
- occupied[pos+2] = true;
- occupied[pos+3] = true;
+ occupied[pos + 1] = true;
+ occupied[pos + 2] = true;
+ occupied[pos + 3] = true;
}
assertEquals(random.nextFloat(), segment.getFloat(pos), 0.0);
@@ -1170,19 +1173,18 @@ public abstract class MemorySegmentTestBase {
int pos = random.nextInt(pageSize - 7);
if (occupied[pos] || occupied[pos + 1] || occupied[pos + 2] || occupied[pos + 3] ||
- occupied[pos + 4] || occupied[pos + 5] || occupied[pos + 6] || occupied[pos + 7])
- {
+ occupied[pos + 4] || occupied[pos + 5] || occupied[pos + 6] || occupied[pos + 7]) {
continue;
}
else {
occupied[pos] = true;
- occupied[pos+1] = true;
- occupied[pos+2] = true;
- occupied[pos+3] = true;
- occupied[pos+4] = true;
- occupied[pos+5] = true;
- occupied[pos+6] = true;
- occupied[pos+7] = true;
+ occupied[pos + 1] = true;
+ occupied[pos + 2] = true;
+ occupied[pos + 3] = true;
+ occupied[pos + 4] = true;
+ occupied[pos + 5] = true;
+ occupied[pos + 6] = true;
+ occupied[pos + 7] = true;
}
segment.putDouble(pos, random.nextDouble());
@@ -1195,19 +1197,18 @@ public abstract class MemorySegmentTestBase {
int pos = random.nextInt(pageSize - 7);
if (occupied[pos] || occupied[pos + 1] || occupied[pos + 2] || occupied[pos + 3] ||
- occupied[pos + 4] || occupied[pos + 5] || occupied[pos + 6] || occupied[pos + 7])
- {
+ occupied[pos + 4] || occupied[pos + 5] || occupied[pos + 6] || occupied[pos + 7]) {
continue;
}
else {
occupied[pos] = true;
- occupied[pos+1] = true;
- occupied[pos+2] = true;
- occupied[pos+3] = true;
- occupied[pos+4] = true;
- occupied[pos+5] = true;
- occupied[pos+6] = true;
- occupied[pos+7] = true;
+ occupied[pos + 1] = true;
+ occupied[pos + 2] = true;
+ occupied[pos + 3] = true;
+ occupied[pos + 4] = true;
+ occupied[pos + 5] = true;
+ occupied[pos + 6] = true;
+ occupied[pos + 7] = true;
}
assertEquals(random.nextDouble(), segment.getDouble(pos), 0.0);
@@ -1222,7 +1223,7 @@ public abstract class MemorySegmentTestBase {
public void testBulkBytePutExceptions() {
final MemorySegment segment = createSegment(pageSize);
- byte[] bytes = new byte[pageSize / 4 + (pageSize%4)];
+ byte[] bytes = new byte[pageSize / 4 + (pageSize % 4)];
random.nextBytes(bytes);
// wrong positions into memory segment
@@ -1884,7 +1885,7 @@ public abstract class MemorySegmentTestBase {
testByteBufferGet(false);
testByteBufferGet(true);
}
-
+
private void testByteBufferGet(boolean directBuffer) {
MemorySegment seg = createSegment(pageSize);
byte[] bytes = new byte[pageSize];
@@ -1926,7 +1927,7 @@ public abstract class MemorySegmentTestBase {
ByteBuffer source = directBuffer ?
ByteBuffer.allocateDirect(pageSize) :
ByteBuffer.allocate(pageSize);
-
+
source.put(bytes);
source.clear();
@@ -1969,9 +1970,9 @@ public abstract class MemorySegmentTestBase {
ByteBuffer target = directBuffer ?
ByteBuffer.allocateDirect(pageSize + 49) :
ByteBuffer.allocate(pageSize + 49);
-
+
target.position(19).limit(19 + pageSize);
-
+
ByteBuffer slicedTarget = target.slice();
// transfer the segment in chunks into the byte buffer
@@ -2004,7 +2005,7 @@ public abstract class MemorySegmentTestBase {
ByteBuffer source = directBuffer ?
ByteBuffer.allocateDirect(pageSize + 49) :
ByteBuffer.allocate(pageSize + 49);
-
+
source.put(bytes);
source.position(19).limit(19 + pageSize);
ByteBuffer slicedSource = source.slice();
@@ -2048,8 +2049,7 @@ public abstract class MemorySegmentTestBase {
for (ByteBuffer bb : new ByteBuffer[] {
ByteBuffer.allocate(bbCapacity),
- ByteBuffer.allocateDirect(bbCapacity) } )
- {
+ ByteBuffer.allocateDirect(bbCapacity) }) {
for (int off : validOffsets) {
for (int len : invalidLengths) {
try {
@@ -2141,7 +2141,6 @@ public abstract class MemorySegmentTestBase {
assertEquals(0, bb.position());
assertEquals(bb.capacity(), bb.limit());
-
int pos = bb.capacity() / 3;
int limit = 2 * bb.capacity() / 3;
bb.limit(limit);
@@ -2167,7 +2166,7 @@ public abstract class MemorySegmentTestBase {
assertEquals(pos, bb.position());
assertEquals(limit, bb.limit());
}
-
+
// ------------------------------------------------------------------------
// Comparing and swapping
// ------------------------------------------------------------------------
@@ -2199,7 +2198,7 @@ public abstract class MemorySegmentTestBase {
int pos2 = random.nextInt(bytes2.length);
int len = Math.min(Math.min(bytes1.length - pos1, bytes2.length - pos2),
- random.nextInt(pageSize / 50 ));
+ random.nextInt(pageSize / 50));
int cmp = seg1.compare(seg2, pos1, pos2, len);
@@ -2214,34 +2213,34 @@ public abstract class MemorySegmentTestBase {
@Test
public void testSwapBytes() {
- final int HALF_SIZE = pageSize / 2;
+ final int halfPageSize = pageSize / 2;
final byte[] bytes1 = new byte[pageSize];
- final byte[] bytes2 = new byte[HALF_SIZE];
+ final byte[] bytes2 = new byte[halfPageSize];
Arrays.fill(bytes2, (byte) 1);
MemorySegment seg1 = createSegment(pageSize);
- MemorySegment seg2 = createSegment(HALF_SIZE);
+ MemorySegment seg2 = createSegment(halfPageSize);
seg1.put(0, bytes1);
seg2.put(0, bytes2);
// wap the second half of the first segment with the second segment
int pos = 0;
- while (pos < HALF_SIZE) {
+ while (pos < halfPageSize) {
int len = random.nextInt(pageSize / 40);
- len = Math.min(len, HALF_SIZE - pos);
- seg1.swapBytes(new byte[len], seg2, pos + HALF_SIZE, pos, len);
+ len = Math.min(len, halfPageSize - pos);
+ seg1.swapBytes(new byte[len], seg2, pos + halfPageSize, pos, len);
pos += len;
}
// the second segment should now be all zeros, the first segment should have one in its second half
- for (int i = 0; i < HALF_SIZE; i++) {
+ for (int i = 0; i < halfPageSize; i++) {
assertEquals((byte) 0, seg1.get(i));
assertEquals((byte) 0, seg2.get(i));
- assertEquals((byte) 1, seg1.get(i + HALF_SIZE));
+ assertEquals((byte) 1, seg1.get(i + halfPageSize));
}
}
@@ -2300,9 +2299,9 @@ public abstract class MemorySegmentTestBase {
assertTrue(e instanceof IllegalStateException || e instanceof NullPointerException);
}
}
-
+
// ------------------------------------------------------------------------
- // Miscellaneous
+ // Miscellaneous
// ------------------------------------------------------------------------
@Test
@@ -2389,26 +2388,26 @@ public abstract class MemorySegmentTestBase {
@Test
public void testSizeAndFreeing() {
// a segment without an owner has a null owner
- final int SIZE = 651;
- MemorySegment seg = createSegment(SIZE);
+ final int segmentSize = 651;
+ MemorySegment seg = createSegment(segmentSize);
- assertEquals(SIZE, seg.size());
+ assertEquals(segmentSize, seg.size());
assertFalse(seg.isFreed());
seg.free();
assertTrue(seg.isFreed());
- assertEquals(SIZE, seg.size());
+ assertEquals(segmentSize, seg.size());
}
-
+
// ------------------------------------------------------------------------
// Parametrization to run with different segment sizes
// ------------------------------------------------------------------------
-
+
@Parameterized.Parameters(name = "segment-size = {0}")
public static Collection<Object[]> executionModes(){
return Arrays.asList(
- new Object[] { 32*1024 },
- new Object[] { 4*1024 },
- new Object[] { 512*1024 } );
+ new Object[] { 32 * 1024 },
+ new Object[] { 4 * 1024 },
+ new Object[] { 512 * 1024 });
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentUndersizedTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentUndersizedTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentUndersizedTest.java
index 950da0e..5f80742 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentUndersizedTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentUndersizedTest.java
@@ -31,8 +31,12 @@ import java.nio.ByteBuffer;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+/**
+ * Tests for undersized {@link HeapMemorySegment} and {@link HybridMemorySegment} (in both heap and
+ * off-heap modes).
+ */
public class MemorySegmentUndersizedTest {
-
+
@Test
public void testZeroSizeHeapSegment() {
MemorySegment segment = new HeapMemorySegment(new byte[0]);
@@ -109,7 +113,7 @@ public class MemorySegmentUndersizedTest {
assertTrue(e instanceof IndexOutOfBoundsException);
}
}
-
+
private static void testSegmentWithSizeLargerZero(MemorySegment segment) {
// ------ bytes ------
@@ -145,7 +149,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.put(Integer.MAX_VALUE, (byte) 0);
fail("IndexOutOfBoundsException expected");
@@ -161,7 +165,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.get(1);
fail("IndexOutOfBoundsException expected");
@@ -169,7 +173,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.get(-1);
fail("IndexOutOfBoundsException expected");
@@ -193,7 +197,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.get(Integer.MAX_VALUE);
fail("IndexOutOfBoundsException expected");
@@ -243,7 +247,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.putBoolean(Integer.MAX_VALUE, true);
fail("IndexOutOfBoundsException expected");
@@ -259,7 +263,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.getBoolean(1);
fail("IndexOutOfBoundsException expected");
@@ -373,7 +377,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.getChar(1);
fail("IndexOutOfBoundsException expected");
@@ -487,7 +491,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.getShort(1);
fail("IndexOutOfBoundsException expected");
@@ -971,7 +975,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.getDouble(Integer.MIN_VALUE);
fail("IndexOutOfBoundsException expected");
@@ -979,7 +983,6 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
// ------ byte[] ------
@@ -1046,7 +1049,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.get(1, new byte[7]);
fail("IndexOutOfBoundsException expected");
@@ -1098,8 +1101,8 @@ public class MemorySegmentUndersizedTest {
// ------ ByteBuffer ------
final ByteBuffer buf = ByteBuffer.allocate(7);
- final int numBytes = 3;
-
+ final int numBytes = 3;
+
try {
segment.put(0, buf, numBytes);
fail("IndexOutOfBoundsException expected");
@@ -1139,7 +1142,7 @@ public class MemorySegmentUndersizedTest {
catch (Exception e) {
assertTrue(e instanceof IndexOutOfBoundsException);
}
-
+
try {
segment.put(Integer.MAX_VALUE, buf, numBytes);
fail("IndexOutOfBoundsException expected");
@@ -1215,7 +1218,7 @@ public class MemorySegmentUndersizedTest {
final DataInput dataInput = new DataInputStream(new ByteArrayInputStream(new byte[20]));
final DataOutput dataOutput = new DataOutputStream(new ByteArrayOutputStream());
-
+
try {
segment.put(dataInput, 0, numBytes);
fail("IndexOutOfBoundsException expected");
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-core/src/test/java/org/apache/flink/core/memory/OperationsOnFreedSegmentTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/OperationsOnFreedSegmentTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/OperationsOnFreedSegmentTest.java
index 57f3db1..ad9d63f 100644
--- a/flink-core/src/test/java/org/apache/flink/core/memory/OperationsOnFreedSegmentTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/memory/OperationsOnFreedSegmentTest.java
@@ -28,12 +28,17 @@ import java.io.DataOutput;
import java.io.DataOutputStream;
import java.nio.ByteBuffer;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+/**
+ * Various tests with freed memory segments for {@link HeapMemorySegment} and {@link
+ * HybridMemorySegment} (in both heap and off-heap modes).
+ */
public class OperationsOnFreedSegmentTest {
private static final int PAGE_SIZE = (int) ((Math.random() * 10000) + 1000);
-
+
@Test
public void testSingleSegmentOperationsHeapSegment() throws Exception {
testOpsOnFreedSegment(new HeapMemorySegment(new byte[PAGE_SIZE]));
@@ -78,7 +83,7 @@ public class OperationsOnFreedSegmentTest {
}
}
}
-
+
@Test
public void testCopyTo() {
MemorySegment aliveHeap = new HeapMemorySegment(new byte[PAGE_SIZE]);
@@ -154,13 +159,13 @@ public class OperationsOnFreedSegmentTest {
}
}
}
-
+
private void testOpsOnFreedSegment(MemorySegment segment) throws Exception {
segment.free();
assertTrue(segment.isFreed());
-
- // --------- bytes -----------
-
+
+ // --------- bytes -----------
+
try {
segment.get(0);
fail("Should fail with an exception");
@@ -190,7 +195,7 @@ public class OperationsOnFreedSegmentTest {
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
-
+
try {
segment.get(Integer.MAX_VALUE);
fail("Should fail with an exception");
@@ -245,7 +250,7 @@ public class OperationsOnFreedSegmentTest {
}
catch (IllegalStateException | NullPointerException ignored) {}
- // --------- booleans -----------
+ // --------- booleans -----------
try {
segment.getBoolean(0);
@@ -330,8 +335,8 @@ public class OperationsOnFreedSegmentTest {
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException ignored) {}
-
- // --------- char -----------
+
+ // --------- char -----------
try {
segment.getChar(0);
@@ -417,7 +422,7 @@ public class OperationsOnFreedSegmentTest {
}
catch (IllegalStateException ignored) {}
- // --------- short -----------
+ // --------- short -----------
try {
segment.getShort(0);
@@ -503,7 +508,7 @@ public class OperationsOnFreedSegmentTest {
}
catch (IllegalStateException ignored) {}
- // --------- integer -----------
+ // --------- integer -----------
try {
segment.getInt(0);
@@ -589,7 +594,7 @@ public class OperationsOnFreedSegmentTest {
}
catch (IllegalStateException ignored) {}
- // --------- longs -----------
+ // --------- longs -----------
try {
segment.getLong(0);
@@ -675,7 +680,7 @@ public class OperationsOnFreedSegmentTest {
}
catch (IllegalStateException ignored) {}
- // --------- floats -----------
+ // --------- floats -----------
try {
segment.getFloat(0);
@@ -761,7 +766,7 @@ public class OperationsOnFreedSegmentTest {
}
catch (IllegalStateException ignored) {}
- // --------- doubles -----------
+ // --------- doubles -----------
try {
segment.getDouble(0);
@@ -847,10 +852,10 @@ public class OperationsOnFreedSegmentTest {
}
catch (IllegalStateException ignored) {}
- // --------- byte[] -----------
+ // --------- byte[] -----------
final byte[] array = new byte[55];
-
+
try {
segment.get(0, array, 3, 17);
fail("Should fail with an exception");
@@ -935,90 +940,90 @@ public class OperationsOnFreedSegmentTest {
}
catch (IllegalStateException | NullPointerException ignored) {}
- // --------- ByteBuffer -----------
+ // --------- ByteBuffer -----------
for (ByteBuffer bbuf : new ByteBuffer[] {
ByteBuffer.allocate(55),
- ByteBuffer.allocateDirect(55) } )
- {
+ ByteBuffer.allocateDirect(55) }) {
+
try {
segment.get(0, bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException ignored) {}
-
+
try {
segment.get(-1, bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
-
+
try {
segment.get(1, bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException ignored) {}
-
+
try {
segment.get(segment.size(), bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException ignored) {}
-
+
try {
segment.get(-segment.size(), bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
-
+
try {
segment.get(Integer.MAX_VALUE, bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
-
+
try {
segment.get(Integer.MIN_VALUE, bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
-
+
try {
segment.put(0, bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
-
+
try {
segment.put(-1, bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
-
+
try {
segment.put(1, bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException ignored) {}
-
+
try {
segment.put(segment.size(), bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException ignored) {}
-
+
try {
segment.put(-segment.size(), bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
-
+
try {
- segment.put(Integer.MAX_VALUE,bbuf, 17);
+ segment.put(Integer.MAX_VALUE, bbuf, 17);
fail("Should fail with an exception");
}
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
-
+
try {
segment.put(Integer.MIN_VALUE, bbuf, 17);
fail("Should fail with an exception");
@@ -1026,7 +1031,7 @@ public class OperationsOnFreedSegmentTest {
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
}
- // --------- Data Input / Output -----------
+ // --------- Data Input / Output -----------
final DataInput din = new DataInputStream(new ByteArrayInputStream(new byte[100]));
final DataOutput dout = new DataOutputStream(new ByteArrayOutputStream());
@@ -1115,12 +1120,11 @@ public class OperationsOnFreedSegmentTest {
}
catch (IllegalStateException | NullPointerException | IndexOutOfBoundsException ignored) {}
}
-
-
+
private void testCompare(MemorySegment seg1, MemorySegment seg2) {
int[] offsetsToTest = { 0, 1, -1, seg1.size(), -seg1.size(), Integer.MAX_VALUE, Integer.MIN_VALUE };
int[] lengthsToTest = { 1, seg1.size(), Integer.MAX_VALUE };
-
+
for (int off1 : offsetsToTest) {
for (int off2 : offsetsToTest) {
for (int len : lengthsToTest) {
@@ -1155,7 +1159,7 @@ public class OperationsOnFreedSegmentTest {
int[] offsetsToTest = { 0, 1, -1, seg1.size(), -seg1.size(), Integer.MAX_VALUE, Integer.MIN_VALUE };
int[] lengthsToTest = { 0, 1, -1, seg1.size(), -seg1.size(), Integer.MAX_VALUE, Integer.MIN_VALUE };
byte[] swapBuffer = new byte[seg1.size()];
-
+
for (int off1 : offsetsToTest) {
for (int off2 : offsetsToTest) {
for (int len : lengthsToTest) {
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
index e53237a..192a586 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
@@ -685,7 +685,7 @@ public class MemoryManager {
hybridSegment.free();
}
else {
- throw new IllegalArgumentException("Memory segment is not a " + HeapMemorySegment.class.getSimpleName());
+ throw new IllegalArgumentException("Memory segment is not a " + HybridMemorySegment.class.getSimpleName());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c3235c39/tools/maven/suppressions-core.xml
----------------------------------------------------------------------
diff --git a/tools/maven/suppressions-core.xml b/tools/maven/suppressions-core.xml
index d0d838a..066b30a 100644
--- a/tools/maven/suppressions-core.xml
+++ b/tools/maven/suppressions-core.xml
@@ -128,14 +128,6 @@ under the License.
checks="AvoidStarImport"/>
<suppress
- files="(.*)core[/\\]memory[/\\](.*)"
- checks="NewlineAtEndOfFile|RegexpSingleline|TodoComment|RedundantImport|ImportOrder|RedundantModifier|JavadocMethod|JavadocParagraph|JavadocType|JavadocStyle|PackageName|TypeNameCheck|ConstantNameCheck|StaticVariableNameCheck|MemberNameCheck|MethodNameCheck|ParameterName|LocalFinalVariableName|LocalVariableName|LeftCurly|UpperEll|FallThrough|reliefPattern|SimplifyBooleanExpression|EmptyStatement|ModifierOrder|EmptyLineSeparator|WhitespaceAround|WhitespaceAfter|NoWhitespaceAfter|NoWhitespaceBefore|OperatorWrap|ParenPad"/>
- <!--Only additional checks for test sources. Those checks were present in the "pre-strict" checkstyle but were not applied to test sources. We do not want to suppress them for sources directory-->
- <suppress
- files="(.*)test[/\\](.*)core[/\\]memory[/\\](.*)"
- checks="AvoidStarImport"/>
-
- <suppress
files="(.*)migration[/\\](.*)"
checks="NewlineAtEndOfFile|RegexpSingleline|TodoComment|RedundantImport|ImportOrder|RedundantModifier|JavadocMethod|JavadocParagraph|JavadocType|JavadocStyle|PackageName|TypeNameCheck|ConstantNameCheck|StaticVariableNameCheck|MemberNameCheck|MethodNameCheck|ParameterName|LocalFinalVariableName|LocalVariableName|LeftCurly|UpperEll|FallThrough|reliefPattern|SimplifyBooleanExpression|EmptyStatement|ModifierOrder|EmptyLineSeparator|WhitespaceAround|WhitespaceAfter|NoWhitespaceAfter|NoWhitespaceBefore|OperatorWrap|ParenPad"/>
[07/10] flink git commit: [hotfix] [build] Remove incorrect and
ineffective example exclusions
Posted by se...@apache.org.
[hotfix] [build] Remove incorrect and ineffective example exclusions
The examples are not part of dist's dependencies, hence no need to exclude them
from the fat jar.
The exclusion did not work anyways, because it used wrong artifact names
(not using scala suffixes).
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8e02df0b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8e02df0b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8e02df0b
Branch: refs/heads/master
Commit: 8e02df0b6c4e52ae4b27bf8d5234770d3b0f678d
Parents: 774e7f7
Author: Stephan Ewen <se...@apache.org>
Authored: Fri Oct 6 13:31:22 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 20:12:51 2017 +0200
----------------------------------------------------------------------
flink-dist/pom.xml | 3 ---
1 file changed, 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8e02df0b/flink-dist/pom.xml
----------------------------------------------------------------------
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index 0490dcc..b874c49 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -503,9 +503,6 @@ under the License.
</filters>
<artifactSet>
<excludes>
- <exclude>org.apache.flink:flink-examples-batch</exclude>
- <exclude>org.apache.flink:flink-examples-streaming</exclude>
- <exclude>org.apache.flink:flink-examples-table</exclude>
<exclude>org.apache.flink:flink-python_${scala.binary.version}</exclude>
<exclude>org.slf4j:slf4j-log4j12</exclude>
<exclude>log4j:log4j</exclude>
[10/10] flink git commit: [FLINK-7310] [core] always use the
HybridMemorySegment
Posted by se...@apache.org.
[FLINK-7310] [core] always use the HybridMemorySegment
Since we'd like to use our own off-heap buffers for network communication, we
cannot use HeapMemorySegment anymore and need to rely on HybridMemorySegment.
We thus drop any code that loads the HeapMemorySegment (it is still available
if needed) in favour of the HybridMemorySegment which is able to work on both
heap and off-heap memory.
For the performance penalty of this change compared to using HeapMemorySegment
alone, see this interesting blob article (from 2015):
https://flink.apache.org/news/2015/09/16/off-heap-memory.html
This closes #4445
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e14caef3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e14caef3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e14caef3
Branch: refs/heads/master
Commit: e14caef3794d0acde4bc86f67eedd8d032a3f213
Parents: c3235c3
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Wed Aug 2 11:34:54 2017 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 6 20:57:07 2017 +0200
----------------------------------------------------------------------
.../flink/core/memory/HeapMemorySegment.java | 36 +++--
.../flink/core/memory/HybridMemorySegment.java | 53 ++-----
.../flink/core/memory/MemorySegmentFactory.java | 144 ++-----------------
.../core/memory/MemorySegmentFactoryTest.java | 67 ---------
.../io/network/buffer/NetworkBufferPool.java | 2 +-
.../flink/runtime/memory/MemoryManager.java | 40 +++---
.../TaskManagerServicesConfiguration.java | 17 ---
.../PartitionRequestClientHandlerTest.java | 4 +-
.../operators/util/HashVsSortMiniBenchmark.java | 30 ++++
9 files changed, 99 insertions(+), 294 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java b/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
index 9aa81d0..96c013a 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/HeapMemorySegment.java
@@ -37,6 +37,7 @@ import java.util.Objects;
* <p>Note that memory segments should usually not be allocated manually, but rather through the
* {@link MemorySegmentFactory}.
*/
+@SuppressWarnings("unused")
@Internal
public final class HeapMemorySegment extends MemorySegment {
@@ -177,29 +178,44 @@ public final class HeapMemorySegment extends MemorySegment {
* A memory segment factory that produces heap memory segments. Note that this factory does not
* support to allocate off-heap memory.
*/
- public static final class HeapMemorySegmentFactory implements MemorySegmentFactory.Factory {
+ public static final class HeapMemorySegmentFactory {
- @Override
+ /**
+ * Creates a new memory segment that targets the given heap memory region.
+ *
+ * @param memory The heap memory region.
+ * @return A new memory segment that targets the given heap memory region.
+ */
public HeapMemorySegment wrap(byte[] memory) {
return new HeapMemorySegment(memory);
}
- @Override
+ /**
+ * Allocates some unpooled memory and creates a new memory segment that represents
+ * that memory.
+ *
+ * @param size The size of the memory segment to allocate.
+ * @param owner The owner to associate with the memory segment.
+ * @return A new memory segment, backed by unpooled heap memory.
+ */
public HeapMemorySegment allocateUnpooledSegment(int size, Object owner) {
return new HeapMemorySegment(new byte[size], owner);
}
- @Override
+ /**
+ * Creates a memory segment that wraps the given byte array.
+ *
+ * <p>This method is intended to be used for components which pool memory and create
+ * memory segments around long-lived memory regions.
+ *
+ * @param memory The heap memory to be represented by the memory segment.
+ * @param owner The owner to associate with the memory segment.
+ * @return A new memory segment representing the given heap memory.
+ */
public HeapMemorySegment wrapPooledHeapMemory(byte[] memory, Object owner) {
return new HeapMemorySegment(memory, owner);
}
- @Override
- public HeapMemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner) {
- throw new UnsupportedOperationException(
- "The MemorySegment factory was not initialized for off-heap memory.");
- }
-
/**
* Prevent external instantiation.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java b/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
index 0b23ef3..a2cf3d3 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
@@ -158,7 +158,7 @@ public final class HybridMemorySegment extends MemorySegment {
// ------------------------------------------------------------------------
@Override
- public byte get(int index) {
+ public final byte get(int index) {
final long pos = address + index;
if (index >= 0 && pos < addressLimit) {
return UNSAFE.getByte(heapMemory, pos);
@@ -173,7 +173,7 @@ public final class HybridMemorySegment extends MemorySegment {
}
@Override
- public void put(int index, byte b) {
+ public final void put(int index, byte b) {
final long pos = address + index;
if (index >= 0 && pos < addressLimit) {
UNSAFE.putByte(heapMemory, pos, b);
@@ -188,17 +188,17 @@ public final class HybridMemorySegment extends MemorySegment {
}
@Override
- public void get(int index, byte[] dst) {
+ public final void get(int index, byte[] dst) {
get(index, dst, 0, dst.length);
}
@Override
- public void put(int index, byte[] src) {
+ public final void put(int index, byte[] src) {
put(index, src, 0, src.length);
}
@Override
- public void get(int index, byte[] dst, int offset, int length) {
+ public final void get(int index, byte[] dst, int offset, int length) {
// check the byte array offset and length and the status
if ((offset | length | (offset + length) | (dst.length - (offset + length))) < 0) {
throw new IndexOutOfBoundsException();
@@ -219,7 +219,7 @@ public final class HybridMemorySegment extends MemorySegment {
}
@Override
- public void put(int index, byte[] src, int offset, int length) {
+ public final void put(int index, byte[] src, int offset, int length) {
// check the byte array offset and length
if ((offset | length | (offset + length) | (src.length - (offset + length))) < 0) {
throw new IndexOutOfBoundsException();
@@ -241,12 +241,12 @@ public final class HybridMemorySegment extends MemorySegment {
}
@Override
- public boolean getBoolean(int index) {
+ public final boolean getBoolean(int index) {
return get(index) != 0;
}
@Override
- public void putBoolean(int index, boolean value) {
+ public final void putBoolean(int index, boolean value) {
put(index, (byte) (value ? 1 : 0));
}
@@ -436,41 +436,4 @@ public final class HybridMemorySegment extends MemorySegment {
}
return getAddress(buffer);
}
-
- // -------------------------------------------------------------------------
- // Factoring
- // -------------------------------------------------------------------------
-
- /**
- * Base factory for hybrid memory segments.
- */
- public static final class HybridMemorySegmentFactory implements MemorySegmentFactory.Factory {
-
- @Override
- public HybridMemorySegment wrap(byte[] memory) {
- return new HybridMemorySegment(memory);
- }
-
- @Override
- public HybridMemorySegment allocateUnpooledSegment(int size, Object owner) {
- return new HybridMemorySegment(new byte[size], owner);
- }
-
- @Override
- public HybridMemorySegment wrapPooledHeapMemory(byte[] memory, Object owner) {
- return new HybridMemorySegment(memory, owner);
- }
-
- @Override
- public HybridMemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner) {
- return new HybridMemorySegment(memory, owner);
- }
-
- /**
- * Prevent external instantiation.
- */
- HybridMemorySegmentFactory() {}
- }
-
- public static final HybridMemorySegmentFactory FACTORY = new HybridMemorySegmentFactory();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
index 5411601..c235999 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
@@ -22,35 +22,27 @@ import org.apache.flink.annotation.Internal;
import java.nio.ByteBuffer;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
/**
- * A factory for memory segments. The purpose of this factory is to make sure that all memory segments
- * for heap data are of the same type. That way, the runtime does not mix the various specializations
- * of the {@link org.apache.flink.core.memory.MemorySegment}. Not mixing them has shown to be beneficial
- * to method specialization by the JIT and to overall performance.
+ * A factory for (hybrid) memory segments ({@link HybridMemorySegment}).
*
- * <p>Note that this factory auto-initialized to use {@link org.apache.flink.core.memory.HeapMemorySegment},
- * if a request to create a segment comes before the initialization.
+ * <p>The purpose of this factory is to make sure that all memory segments for heap data are of the
+ * same type. That way, the runtime does not mix the various specializations of the {@link
+ * MemorySegment}. Not mixing them has shown to be beneficial to method specialization by the JIT
+ * and to overall performance.
*/
@Internal
-public class MemorySegmentFactory {
-
- /**
- * The factory to use.
- */
- private static volatile Factory factory;
+public final class MemorySegmentFactory {
/**
* Creates a new memory segment that targets the given heap memory region.
- * This method should be used to turn short lived byte arrays into memory segments.
+ *
+ * <p>This method should be used to turn short lived byte arrays into memory segments.
*
* @param buffer The heap memory region.
* @return A new memory segment that targets the given heap memory region.
*/
public static MemorySegment wrap(byte[] buffer) {
- ensureInitialized();
- return factory.wrap(buffer);
+ return new HybridMemorySegment(buffer);
}
/**
@@ -79,8 +71,7 @@ public class MemorySegmentFactory {
* @return A new memory segment, backed by unpooled heap memory.
*/
public static MemorySegment allocateUnpooledSegment(int size, Object owner) {
- ensureInitialized();
- return factory.allocateUnpooledSegment(size, owner);
+ return new HybridMemorySegment(new byte[size], owner);
}
/**
@@ -94,8 +85,7 @@ public class MemorySegmentFactory {
* @return A new memory segment representing the given heap memory.
*/
public static MemorySegment wrapPooledHeapMemory(byte[] memory, Object owner) {
- ensureInitialized();
- return factory.wrapPooledHeapMemory(memory, owner);
+ return new HybridMemorySegment(memory, owner);
}
/**
@@ -110,117 +100,7 @@ public class MemorySegmentFactory {
* @return A new memory segment representing the given off-heap memory.
*/
public static MemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner) {
- ensureInitialized();
- return factory.wrapPooledOffHeapMemory(memory, owner);
- }
-
- // ------------------------------------------------------------------------
-
- /**
- * Initializes this factory with the given concrete factory, iff it is not yet initialized.
- * This also checks if the factory is already initialized to the exact same concrete factory
- * as given.
- *
- * @param f The concrete factory to use.
- * @return True, if the factory is initialized with the given concrete factory, or if it was already
- * initialized with the exact same concrete factory. False, if it is already initialized with
- * a different concrete factory.
- */
- public static boolean initializeIfNotInitialized(Factory f) {
- checkNotNull(f);
-
- synchronized (MemorySegmentFactory.class) {
- if (factory == null) {
- factory = f;
- return true;
- } else {
- return factory == f;
- }
- }
- }
-
- /**
- * Checks whether this memory segment factory has been initialized (with a type to produce).
- *
- * @return True, if the factory has been initialized, false otherwise.
- */
- public static boolean isInitialized() {
- return factory != null;
+ return new HybridMemorySegment(memory, owner);
}
- /**
- * Gets the factory. May return null, if the factory has not been initialized.
- *
- * @return The factory, or null, if the factory has not been initialized.
- */
- public static Factory getFactory() {
- return factory;
- }
-
- /**
- * Sets the factory to the {@link HeapMemorySegment#FACTORY} if no factory has been initialized, yet.
- */
- private static void ensureInitialized() {
- if (factory == null) {
- synchronized (MemorySegmentFactory.class) {
- if (factory == null) {
- factory = HeapMemorySegment.FACTORY;
- }
- }
- }
- }
-
- // ------------------------------------------------------------------------
- // Internal factory
- // ------------------------------------------------------------------------
-
- /**
- * A concrete factory for memory segments.
- */
- public interface Factory {
-
- /**
- * Creates a new memory segment that targets the given heap memory region.
- *
- * @param memory The heap memory region.
- * @return A new memory segment that targets the given heap memory region.
- */
- MemorySegment wrap(byte[] memory);
-
- /**
- * Allocates some unpooled memory and creates a new memory segment that represents
- * that memory.
- *
- * @param size The size of the memory segment to allocate.
- * @param owner The owner to associate with the memory segment.
- * @return A new memory segment, backed by unpooled heap memory.
- */
- MemorySegment allocateUnpooledSegment(int size, Object owner);
-
- /**
- * Creates a memory segment that wraps the given byte array.
- *
- * <p>This method is intended to be used for components which pool memory and create
- * memory segments around long-lived memory regions.
- *
- *
- * @param memory The heap memory to be represented by the memory segment.
- * @param owner The owner to associate with the memory segment.
- * @return A new memory segment representing the given heap memory.
- */
- MemorySegment wrapPooledHeapMemory(byte[] memory, Object owner);
-
- /**
- * Creates a memory segment that wraps the off-heap memory backing the given ByteBuffer.
- * Note that the ByteBuffer needs to be a <i>direct ByteBuffer</i>.
- *
- * <p>This method is intended to be used for components which pool memory and create
- * memory segments around long-lived memory regions.
- *
- * @param memory The byte buffer with the off-heap memory to be represented by the memory segment.
- * @param owner The owner to associate with the memory segment.
- * @return A new memory segment representing the given off-heap memory.
- */
- MemorySegment wrapPooledOffHeapMemory(ByteBuffer memory, Object owner);
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java b/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java
deleted file mode 100644
index 3fd2cbe..0000000
--- a/flink-core/src/test/java/org/apache/flink/core/memory/MemorySegmentFactoryTest.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.core.memory;
-
-import org.junit.Test;
-
-import java.lang.reflect.Field;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-/**
- * Tests for the {@link MemorySegmentFactory} in on/off-heap modes.
- */
-public class MemorySegmentFactoryTest {
-
- @Test
- public void testInitializationToHeapSegments() throws Exception {
- clearSegmentFactory();
-
- assertFalse(MemorySegmentFactory.isInitialized());
- assertTrue(MemorySegmentFactory.initializeIfNotInitialized(HeapMemorySegment.FACTORY));
- assertTrue(MemorySegmentFactory.isInitialized());
-
- assertTrue(MemorySegmentFactory.initializeIfNotInitialized(HeapMemorySegment.FACTORY));
- assertFalse(MemorySegmentFactory.initializeIfNotInitialized(HybridMemorySegment.FACTORY));
-
- assertEquals(HeapMemorySegment.FACTORY, MemorySegmentFactory.getFactory());
- }
-
- @Test
- public void testInitializationToOffHeapSegments() throws Exception {
- clearSegmentFactory();
-
- assertFalse(MemorySegmentFactory.isInitialized());
- assertTrue(MemorySegmentFactory.initializeIfNotInitialized(HybridMemorySegment.FACTORY));
- assertTrue(MemorySegmentFactory.isInitialized());
-
- assertTrue(MemorySegmentFactory.initializeIfNotInitialized(HybridMemorySegment.FACTORY));
- assertFalse(MemorySegmentFactory.initializeIfNotInitialized(HeapMemorySegment.FACTORY));
-
- assertEquals(HybridMemorySegment.FACTORY, MemorySegmentFactory.getFactory());
- }
-
- private static void clearSegmentFactory() throws Exception {
- Field factoryField = MemorySegmentFactory.class.getDeclaredField("factory");
- factoryField.setAccessible(true);
- factoryField.set(null, null);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
index 1eb44c3..b70e912 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
@@ -74,7 +74,7 @@ public class NetworkBufferPool implements BufferPoolFactory {
final long sizeInLong = (long) segmentSize;
try {
- this.availableMemorySegments = new ArrayBlockingQueue<MemorySegment>(numberOfSegmentsToAllocate);
+ this.availableMemorySegments = new ArrayBlockingQueue<>(numberOfSegmentsToAllocate);
}
catch (OutOfMemoryError err) {
throw new OutOfMemoryError("Could not allocate buffer queue of length "
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
index 192a586..c1a98cf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
@@ -18,9 +18,9 @@
package org.apache.flink.runtime.memory;
-import org.apache.flink.core.memory.HeapMemorySegment;
import org.apache.flink.core.memory.HybridMemorySegment;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.core.memory.MemoryType;
import org.apache.flink.util.MathUtils;
@@ -43,8 +43,8 @@ import java.util.Set;
* is represented in segments of equal size. Operators allocate the memory by requesting a number
* of memory segments.
*
- * <p>The memory may be represented as on-heap byte arrays ({@link HeapMemorySegment}), or as off-heap
- * memory regions ({@link HybridMemorySegment}). Which kind of memory the MemoryManager serves can
+ * <p>The memory may be represented as on-heap byte arrays or as off-heap memory regions
+ * (both via {@link HybridMemorySegment}). Which kind of memory the MemoryManager serves can
* be passed as an argument to the initialization.
*
* <p>The memory manager can either pre-allocate all memory, or allocate the memory on demand. In the
@@ -164,7 +164,7 @@ public class MemoryManager {
switch (memoryType) {
case HEAP:
- this.memoryPool = new HeapMemoryPool(memToAllocate, pageSize);
+ this.memoryPool = new HybridHeapMemoryPool(memToAllocate, pageSize);
break;
case OFF_HEAP:
if (!preAllocateMemory) {
@@ -598,15 +598,15 @@ public class MemoryManager {
abstract void clear();
}
- static final class HeapMemoryPool extends MemoryPool {
+ static final class HybridHeapMemoryPool extends MemoryPool {
/** The collection of available memory segments. */
private final ArrayDeque<byte[]> availableMemory;
private final int segmentSize;
- public HeapMemoryPool(int numInitialSegments, int segmentSize) {
- this.availableMemory = new ArrayDeque<byte[]>(numInitialSegments);
+ HybridHeapMemoryPool(int numInitialSegments, int segmentSize) {
+ this.availableMemory = new ArrayDeque<>(numInitialSegments);
this.segmentSize = segmentSize;
for (int i = 0; i < numInitialSegments; i++) {
@@ -615,25 +615,25 @@ public class MemoryManager {
}
@Override
- HeapMemorySegment allocateNewSegment(Object owner) {
- return HeapMemorySegment.FACTORY.allocateUnpooledSegment(segmentSize, owner);
+ MemorySegment allocateNewSegment(Object owner) {
+ return MemorySegmentFactory.allocateUnpooledSegment(segmentSize, owner);
}
@Override
- HeapMemorySegment requestSegmentFromPool(Object owner) {
+ MemorySegment requestSegmentFromPool(Object owner) {
byte[] buf = availableMemory.remove();
- return HeapMemorySegment.FACTORY.wrapPooledHeapMemory(buf, owner);
+ return MemorySegmentFactory.wrapPooledHeapMemory(buf, owner);
}
@Override
void returnSegmentToPool(MemorySegment segment) {
- if (segment.getClass() == HeapMemorySegment.class) {
- HeapMemorySegment heapSegment = (HeapMemorySegment) segment;
+ if (segment.getClass() == HybridMemorySegment.class) {
+ HybridMemorySegment heapSegment = (HybridMemorySegment) segment;
availableMemory.add(heapSegment.getArray());
heapSegment.free();
}
else {
- throw new IllegalArgumentException("Memory segment is not a " + HeapMemorySegment.class.getSimpleName());
+ throw new IllegalArgumentException("Memory segment is not a " + HybridMemorySegment.class.getSimpleName());
}
}
@@ -655,8 +655,8 @@ public class MemoryManager {
private final int segmentSize;
- public HybridOffHeapMemoryPool(int numInitialSegments, int segmentSize) {
- this.availableMemory = new ArrayDeque<ByteBuffer>(numInitialSegments);
+ HybridOffHeapMemoryPool(int numInitialSegments, int segmentSize) {
+ this.availableMemory = new ArrayDeque<>(numInitialSegments);
this.segmentSize = segmentSize;
for (int i = 0; i < numInitialSegments; i++) {
@@ -665,15 +665,15 @@ public class MemoryManager {
}
@Override
- HybridMemorySegment allocateNewSegment(Object owner) {
+ MemorySegment allocateNewSegment(Object owner) {
ByteBuffer memory = ByteBuffer.allocateDirect(segmentSize);
- return HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(memory, owner);
+ return MemorySegmentFactory.wrapPooledOffHeapMemory(memory, owner);
}
@Override
- HybridMemorySegment requestSegmentFromPool(Object owner) {
+ MemorySegment requestSegmentFromPool(Object owner) {
ByteBuffer buf = availableMemory.remove();
- return HybridMemorySegment.FACTORY.wrapPooledOffHeapMemory(buf, owner);
+ return MemorySegmentFactory.wrapPooledOffHeapMemory(buf, owner);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
index d1cfc4c..6cc7569 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
@@ -23,9 +23,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.QueryableStateOptions;
import org.apache.flink.configuration.TaskManagerOptions;
-import org.apache.flink.core.memory.HeapMemorySegment;
-import org.apache.flink.core.memory.HybridMemorySegment;
-import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.core.memory.MemoryType;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
@@ -281,20 +278,6 @@ public class TaskManagerServicesConfiguration {
memType = MemoryType.HEAP;
}
- // initialize the memory segment factory accordingly
- // TODO - this should be in the TaskManager, not the configuration
- if (memType == MemoryType.HEAP) {
- if (!MemorySegmentFactory.initializeIfNotInitialized(HeapMemorySegment.FACTORY)) {
- throw new Exception("Memory type is set to heap memory, but memory segment " +
- "factory has been initialized for off-heap memory segments");
- }
- } else {
- if (!MemorySegmentFactory.initializeIfNotInitialized(HybridMemorySegment.FACTORY)) {
- throw new Exception("Memory type is set to off-heap memory, but memory segment " +
- "factory has been initialized for heap memory segments");
- }
- }
-
// network buffer memory fraction
float networkBufFraction = configuration.getFloat(TaskManagerOptions.NETWORK_BUFFERS_MEMORY_FRACTION);
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
index ff19967..f3f6feb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
@@ -18,8 +18,8 @@
package org.apache.flink.runtime.io.network.netty;
-import org.apache.flink.core.memory.HeapMemorySegment;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferProvider;
import org.apache.flink.runtime.io.network.netty.NettyMessage.BufferResponse;
@@ -239,7 +239,7 @@ public class PartitionRequestClientHandlerTest {
// ---------------------------------------------------------------------------------------------
private static Buffer createBuffer(boolean fill) {
- MemorySegment segment = HeapMemorySegment.FACTORY.allocateUnpooledSegment(1024, null);
+ MemorySegment segment = MemorySegmentFactory.allocateUnpooledSegment(1024, null);
if (fill) {
for (int i = 0; i < 1024; i++) {
segment.put(i, (byte) i);
http://git-wip-us.apache.org/repos/asf/flink/blob/e14caef3/flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/HashVsSortMiniBenchmark.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/HashVsSortMiniBenchmark.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/HashVsSortMiniBenchmark.java
index 4da2c8c..2696045 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/HashVsSortMiniBenchmark.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/HashVsSortMiniBenchmark.java
@@ -40,6 +40,7 @@ import org.apache.flink.runtime.operators.testutils.TestData.TupleGenerator.KeyM
import org.apache.flink.runtime.operators.testutils.TestData.TupleGenerator.ValueMode;
import org.apache.flink.util.Collector;
import org.apache.flink.util.MutableObjectIterator;
+
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -244,6 +245,35 @@ public class HashVsSortMiniBenchmark {
}
}
+ @Test
+ public void testSortOnly() throws Exception {
+ TestData.TupleGenerator generator1 = new TestData.TupleGenerator(SEED1, INPUT_1_SIZE / 10, 100, KeyMode.RANDOM, ValueMode.RANDOM_LENGTH);
+
+ final TestData.TupleGeneratorIterator input1 = new TestData.TupleGeneratorIterator(generator1, INPUT_1_SIZE);
+
+ long start = System.nanoTime();
+
+ final UnilateralSortMerger<Tuple2<Integer, String>> sorter = new UnilateralSortMerger<>(
+ this.memoryManager, this.ioManager, input1, this.parentTask, this.serializer1,
+ this.comparator1.duplicate(), (double)MEMORY_FOR_SORTER/MEMORY_SIZE, 128, 0.8f,
+ true /*use large record handler*/, true);
+
+ MutableObjectIterator<Tuple2<Integer, String>> iter = sorter.getIterator();
+
+ long stop1 = System.nanoTime();
+
+ Tuple2<Integer, String> t = new Tuple2<>();
+ while (iter.next() != null);
+
+ long stop2 = System.nanoTime();
+
+ long sortMsecs = (stop1 - start) / 1_000_000;
+ long allMsecs = (stop2 - start) / 1_000_000;
+
+ System.out.printf("Sort only took %d / %d msecs\n", sortMsecs, allMsecs);
+
+ sorter.close();
+ }
private static final class NoOpMatcher implements FlatJoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple2<Integer, String>> {
private static final long serialVersionUID = 1L;