You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2015/01/12 09:16:15 UTC
[07/18] flink git commit: [FLINK-986] [FLINK-25] [Distributed
runtime] Add initial support for intermediate results
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/util/DataInputDeserializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/DataInputDeserializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/DataInputDeserializer.java
new file mode 100644
index 0000000..6952ce8
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/DataInputDeserializer.java
@@ -0,0 +1,362 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.util;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UTFDataFormatException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.MemoryUtils;
+
+/**
+ * A simple and efficient deserializer for the {@link java.io.DataInput} interface.
+ */
+public class DataInputDeserializer implements DataInputView {
+
+ private byte[] buffer;
+
+ private int end;
+
+ private int position;
+
+ public DataInputDeserializer() {
+ }
+
+ public DataInputDeserializer(byte[] buffer, int start, int len) {
+ setBuffer(buffer, start, len);
+ }
+
+ public DataInputDeserializer(ByteBuffer buffer) {
+ setBuffer(buffer);
+ }
+
+ public void setBuffer(ByteBuffer buffer) {
+ if (buffer.hasArray()) {
+ this.buffer = buffer.array();
+ this.position = buffer.arrayOffset() + buffer.position();
+ this.end = this.position + buffer.remaining();
+ } else if (buffer.isDirect()) {
+ this.buffer = new byte[buffer.remaining()];
+ this.position = 0;
+ this.end = this.buffer.length;
+
+ buffer.get(this.buffer);
+ } else {
+ throw new IllegalArgumentException("The given buffer is neither an array-backed heap ByteBuffer, nor a direct ByteBuffer.");
+ }
+ }
+
+ public void setBuffer(byte[] buffer, int start, int len) {
+ if (buffer == null) {
+ throw new NullPointerException();
+ }
+
+ if (start < 0 || len < 0 || start + len >= buffer.length) {
+ throw new IllegalArgumentException();
+ }
+
+ this.buffer = buffer;
+ this.position = start;
+ this.end = start + len;
+ }
+
+ // ----------------------------------------------------------------------------------------
+ // Data Input
+ // ----------------------------------------------------------------------------------------
+
+ @Override
+ public boolean readBoolean() throws IOException {
+ if (this.position < this.end) {
+ return this.buffer[this.position++] != 0;
+ } else {
+ throw new EOFException();
+ }
+ }
+
+ @Override
+ public byte readByte() throws IOException {
+ if (this.position < this.end) {
+ return this.buffer[this.position++];
+ } else {
+ throw new EOFException();
+ }
+ }
+
+ @Override
+ public char readChar() throws IOException {
+ if (this.position < this.end - 1) {
+ return (char) (((this.buffer[this.position++] & 0xff) << 8) | ((this.buffer[this.position++] & 0xff) << 0));
+ } else {
+ throw new EOFException();
+ }
+ }
+
+ @Override
+ public double readDouble() throws IOException {
+ return Double.longBitsToDouble(readLong());
+ }
+
+ @Override
+ public float readFloat() throws IOException {
+ return Float.intBitsToFloat(readInt());
+ }
+
+ @Override
+ public void readFully(byte[] b) throws IOException {
+ readFully(b, 0, b.length);
+ }
+
+ @Override
+ public void readFully(byte[] b, int off, int len) throws IOException {
+ if (len >= 0) {
+ if (off <= b.length - len) {
+ if (this.position <= this.end - len) {
+ System.arraycopy(this.buffer, position, b, off, len);
+ position += len;
+ } else {
+ throw new EOFException();
+ }
+ } else {
+ throw new ArrayIndexOutOfBoundsException();
+ }
+ } else if (len < 0) {
+ throw new IllegalArgumentException("Length may not be negative.");
+ }
+ }
+
+ @Override
+ public int readInt() throws IOException {
+ if (this.position >= 0 && this.position < this.end - 3) {
+ @SuppressWarnings("restriction")
+ int value = UNSAFE.getInt(this.buffer, BASE_OFFSET + this.position);
+ if (LITTLE_ENDIAN) {
+ value = Integer.reverseBytes(value);
+ }
+
+ this.position += 4;
+ return value;
+ } else {
+ throw new EOFException();
+ }
+ }
+
+ @Override
+ public String readLine() throws IOException {
+ if (this.position < this.end) {
+ // read until a newline is found
+ StringBuilder bld = new StringBuilder();
+ char curr = (char) readUnsignedByte();
+ while (position < this.end && curr != '\n') {
+ bld.append(curr);
+ curr = (char) readUnsignedByte();
+ }
+ // trim a trailing carriage return
+ int len = bld.length();
+ if (len > 0 && bld.charAt(len - 1) == '\r') {
+ bld.setLength(len - 1);
+ }
+ String s = bld.toString();
+ bld.setLength(0);
+ return s;
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public long readLong() throws IOException {
+ if (position >= 0 && position < this.end - 7) {
+ @SuppressWarnings("restriction")
+ long value = UNSAFE.getLong(this.buffer, BASE_OFFSET + this.position);
+ if (LITTLE_ENDIAN) {
+ value = Long.reverseBytes(value);
+ }
+ this.position += 8;
+ return value;
+ } else {
+ throw new EOFException();
+ }
+ }
+
+ @Override
+ public short readShort() throws IOException {
+ if (position >= 0 && position < this.end - 1) {
+ return (short) ((((this.buffer[position++]) & 0xff) << 8) | (((this.buffer[position++]) & 0xff) << 0));
+ } else {
+ throw new EOFException();
+ }
+ }
+
+ @Override
+ public String readUTF() throws IOException {
+ int utflen = readUnsignedShort();
+ byte[] bytearr = new byte[utflen];
+ char[] chararr = new char[utflen];
+
+ int c, char2, char3;
+ int count = 0;
+ int chararr_count = 0;
+
+ readFully(bytearr, 0, utflen);
+
+ while (count < utflen) {
+ c = (int) bytearr[count] & 0xff;
+ if (c > 127) {
+ break;
+ }
+ count++;
+ chararr[chararr_count++] = (char) c;
+ }
+
+ while (count < utflen) {
+ c = (int) bytearr[count] & 0xff;
+ switch (c >> 4) {
+ case 0:
+ case 1:
+ case 2:
+ case 3:
+ case 4:
+ case 5:
+ case 6:
+ case 7:
+ /* 0xxxxxxx */
+ count++;
+ chararr[chararr_count++] = (char) c;
+ break;
+ case 12:
+ case 13:
+ /* 110x xxxx 10xx xxxx */
+ count += 2;
+ if (count > utflen) {
+ throw new UTFDataFormatException("malformed input: partial character at end");
+ }
+ char2 = (int) bytearr[count - 1];
+ if ((char2 & 0xC0) != 0x80) {
+ throw new UTFDataFormatException("malformed input around byte " + count);
+ }
+ chararr[chararr_count++] = (char) (((c & 0x1F) << 6) | (char2 & 0x3F));
+ break;
+ case 14:
+ /* 1110 xxxx 10xx xxxx 10xx xxxx */
+ count += 3;
+ if (count > utflen) {
+ throw new UTFDataFormatException("malformed input: partial character at end");
+ }
+ char2 = (int) bytearr[count - 2];
+ char3 = (int) bytearr[count - 1];
+ if (((char2 & 0xC0) != 0x80) || ((char3 & 0xC0) != 0x80)) {
+ throw new UTFDataFormatException("malformed input around byte " + (count - 1));
+ }
+ chararr[chararr_count++] = (char) (((c & 0x0F) << 12) | ((char2 & 0x3F) << 6) | ((char3 & 0x3F) << 0));
+ break;
+ default:
+ /* 10xx xxxx, 1111 xxxx */
+ throw new UTFDataFormatException("malformed input around byte " + count);
+ }
+ }
+ // The number of chars produced may be less than utflen
+ return new String(chararr, 0, chararr_count);
+ }
+
+ @Override
+ public int readUnsignedByte() throws IOException {
+ if (this.position < this.end) {
+ return (this.buffer[this.position++] & 0xff);
+ } else {
+ throw new EOFException();
+ }
+ }
+
+ @Override
+ public int readUnsignedShort() throws IOException {
+ if (this.position < this.end - 1) {
+ return ((this.buffer[this.position++] & 0xff) << 8) | ((this.buffer[this.position++] & 0xff) << 0);
+ } else {
+ throw new EOFException();
+ }
+ }
+
+ @Override
+ public int skipBytes(int n) throws IOException {
+ if (this.position <= this.end - n) {
+ this.position += n;
+ return n;
+ } else {
+ n = this.end - this.position;
+ this.position = this.end;
+ return n;
+ }
+ }
+
+ @SuppressWarnings("restriction")
+ private static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
+
+ @SuppressWarnings("restriction")
+ private static final long BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
+
+ private static final boolean LITTLE_ENDIAN = (MemoryUtils.NATIVE_BYTE_ORDER == ByteOrder.LITTLE_ENDIAN);
+
+ @Override
+ public void skipBytesToRead(int numBytes) throws IOException {
+ int skippedBytes = skipBytes(numBytes);
+
+ if(skippedBytes < numBytes){
+ throw new EOFException("Could not skip " + numBytes +" bytes.");
+ }
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ if(b == null){
+ throw new NullPointerException("Byte array b cannot be null.");
+ }
+
+ if(off < 0){
+ throw new IndexOutOfBoundsException("Offset cannot be negative.");
+ }
+
+ if(len < 0){
+ throw new IndexOutOfBoundsException("Length cannot be negative.");
+ }
+
+ if(b.length - off < len){
+ throw new IndexOutOfBoundsException("Byte array does not provide enough space to store requested data" +
+ ".");
+ }
+
+ if(this.position >= this.end){
+ return -1;
+ }else{
+ int toRead = Math.min(this.end-this.position, len);
+ System.arraycopy(this.buffer,this.position,b,off,toRead);
+ this.position += toRead;
+
+ return toRead;
+ }
+ }
+
+ @Override
+ public int read(byte[] b) throws IOException {
+ return read(b, 0, b.length);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/util/DataOutputSerializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/DataOutputSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/DataOutputSerializer.java
new file mode 100644
index 0000000..adfec50
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/DataOutputSerializer.java
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.util;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UTFDataFormatException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.core.memory.MemoryUtils;
+
+/**
+ * A simple and efficient serializer for the {@link java.io.DataOutput} interface.
+ */
+public class DataOutputSerializer implements DataOutputView {
+
+ private byte[] buffer;
+
+ private int position;
+
+ private ByteBuffer wrapper;
+
+ public DataOutputSerializer(int startSize) {
+ if (startSize < 1) {
+ throw new IllegalArgumentException();
+ }
+
+ this.buffer = new byte[startSize];
+ this.position = 0;
+ this.wrapper = ByteBuffer.wrap(buffer);
+ }
+
+ public ByteBuffer wrapAsByteBuffer() {
+ this.wrapper.position(0);
+ this.wrapper.limit(this.position);
+ return this.wrapper;
+ }
+
+ public byte[] getByteArray() {
+ return buffer;
+ }
+
+ public void clear() {
+ this.position = 0;
+ }
+
+ public int length() {
+ return this.position;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("[pos=%d cap=%d]", this.position, this.buffer.length);
+ }
+
+ // ----------------------------------------------------------------------------------------
+ // Data Output
+ // ----------------------------------------------------------------------------------------
+
+ @Override
+ public void write(int b) throws IOException {
+ if (this.position >= this.buffer.length) {
+ resize(1);
+ }
+ this.buffer[this.position++] = (byte) (b & 0xff);
+ }
+
+ @Override
+ public void write(byte[] b) throws IOException {
+ write(b, 0, b.length);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ if (len < 0 || off > b.length - len) {
+ throw new ArrayIndexOutOfBoundsException();
+ }
+ if (this.position > this.buffer.length - len) {
+ resize(len);
+ }
+ System.arraycopy(b, off, this.buffer, this.position, len);
+ this.position += len;
+ }
+
+ @Override
+ public void writeBoolean(boolean v) throws IOException {
+ write(v ? 1 : 0);
+ }
+
+ @Override
+ public void writeByte(int v) throws IOException {
+ write(v);
+ }
+
+ @Override
+ public void writeBytes(String s) throws IOException {
+ final int sLen = s.length();
+ if (this.position >= this.buffer.length - sLen) {
+ resize(sLen);
+ }
+
+ for (int i = 0; i < sLen; i++) {
+ writeByte(s.charAt(i));
+ }
+ this.position += sLen;
+ }
+
+ @Override
+ public void writeChar(int v) throws IOException {
+ if (this.position >= this.buffer.length - 1) {
+ resize(2);
+ }
+ this.buffer[this.position++] = (byte) (v >> 8);
+ this.buffer[this.position++] = (byte) v;
+ }
+
+ @Override
+ public void writeChars(String s) throws IOException {
+ final int sLen = s.length();
+ if (this.position >= this.buffer.length - 2 * sLen) {
+ resize(2 * sLen);
+ }
+ for (int i = 0; i < sLen; i++) {
+ writeChar(s.charAt(i));
+ }
+ }
+
+ @Override
+ public void writeDouble(double v) throws IOException {
+ writeLong(Double.doubleToLongBits(v));
+ }
+
+ @Override
+ public void writeFloat(float v) throws IOException {
+ writeInt(Float.floatToIntBits(v));
+ }
+
+ @SuppressWarnings("restriction")
+ @Override
+ public void writeInt(int v) throws IOException {
+ if (this.position >= this.buffer.length - 3) {
+ resize(4);
+ }
+ if (LITTLE_ENDIAN) {
+ v = Integer.reverseBytes(v);
+ }
+ UNSAFE.putInt(this.buffer, BASE_OFFSET + this.position, v);
+ this.position += 4;
+ }
+
+ @SuppressWarnings("restriction")
+ @Override
+ public void writeLong(long v) throws IOException {
+ if (this.position >= this.buffer.length - 7) {
+ resize(8);
+ }
+ if (LITTLE_ENDIAN) {
+ v = Long.reverseBytes(v);
+ }
+ UNSAFE.putLong(this.buffer, BASE_OFFSET + this.position, v);
+ this.position += 8;
+ }
+
+ @Override
+ public void writeShort(int v) throws IOException {
+ if (this.position >= this.buffer.length - 1) {
+ resize(2);
+ }
+ this.buffer[this.position++] = (byte) ((v >>> 8) & 0xff);
+ this.buffer[this.position++] = (byte) ((v >>> 0) & 0xff);
+ }
+
+ @Override
+ public void writeUTF(String str) throws IOException {
+ int strlen = str.length();
+ int utflen = 0;
+ int c;
+
+ /* use charAt instead of copying String to char array */
+ for (int i = 0; i < strlen; i++) {
+ c = str.charAt(i);
+ if ((c >= 0x0001) && (c <= 0x007F)) {
+ utflen++;
+ }
+ else if (c > 0x07FF) {
+ utflen += 3;
+ }
+ else {
+ utflen += 2;
+ }
+ }
+
+ if (utflen > 65535) {
+ throw new UTFDataFormatException("Encoded string is too long: " + utflen);
+ }
+ else if (this.position > this.buffer.length - utflen - 2) {
+ resize(utflen + 2);
+ }
+
+ byte[] bytearr = this.buffer;
+ int count = this.position;
+
+ bytearr[count++] = (byte) ((utflen >>> 8) & 0xFF);
+ bytearr[count++] = (byte) ((utflen >>> 0) & 0xFF);
+
+ int i = 0;
+ for (i = 0; i < strlen; i++) {
+ c = str.charAt(i);
+ if (!((c >= 0x0001) && (c <= 0x007F))) {
+ break;
+ }
+ bytearr[count++] = (byte) c;
+ }
+
+ for (; i < strlen; i++) {
+ c = str.charAt(i);
+ if ((c >= 0x0001) && (c <= 0x007F)) {
+ bytearr[count++] = (byte) c;
+
+ }
+ else if (c > 0x07FF) {
+ bytearr[count++] = (byte) (0xE0 | ((c >> 12) & 0x0F));
+ bytearr[count++] = (byte) (0x80 | ((c >> 6) & 0x3F));
+ bytearr[count++] = (byte) (0x80 | ((c >> 0) & 0x3F));
+ }
+ else {
+ bytearr[count++] = (byte) (0xC0 | ((c >> 6) & 0x1F));
+ bytearr[count++] = (byte) (0x80 | ((c >> 0) & 0x3F));
+ }
+ }
+
+ this.position = count;
+ }
+
+
+ private void resize(int minCapacityAdd) throws IOException {
+ try {
+ final int newLen = Math.max(this.buffer.length * 2, this.buffer.length + minCapacityAdd);
+ final byte[] nb = new byte[newLen];
+ System.arraycopy(this.buffer, 0, nb, 0, this.position);
+ this.buffer = nb;
+ this.wrapper = ByteBuffer.wrap(this.buffer);
+ }
+ catch (NegativeArraySizeException nasex) {
+ throw new IOException("Serialization failed because the record length would exceed 2GB (max addressable array size in Java).");
+ }
+ }
+
+ @SuppressWarnings("restriction")
+ private static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
+
+ @SuppressWarnings("restriction")
+ private static final long BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
+
+ private static final boolean LITTLE_ENDIAN = (MemoryUtils.NATIVE_BYTE_ORDER == ByteOrder.LITTLE_ENDIAN);
+
+ @Override
+ public void skipBytesToWrite(int numBytes) throws IOException {
+ if (buffer.length - this.position < numBytes) {
+ throw new EOFException("Could not skip " + numBytes + " bytes.");
+ }
+
+ this.position += numBytes;
+ }
+
+ @Override
+ public void write(DataInputView source, int numBytes) throws IOException {
+ if (buffer.length - this.position < numBytes) {
+ throw new EOFException("Could not write " + numBytes + " bytes. Buffer overflow.");
+ }
+
+ source.read(this.buffer, this.position, numBytes);
+ this.position += numBytes;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/util/LRUCache.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/LRUCache.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/LRUCache.java
new file mode 100644
index 0000000..8a0cb49
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/LRUCache.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.util;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+public class LRUCache<V> {
+
+ private final Deque<V> lruCache = new ArrayDeque<V>();
+
+ public static <V> LRUCache<V> create() {
+ return new LRUCache<V>();
+ }
+
+ public void add(V value) {
+ lruCache.addLast(value);
+ }
+
+ public V getLRU() {
+ V lru = lruCache.pollFirst();
+ if (lru != null) {
+ add(lru);
+ }
+ return lru;
+ }
+
+ public V removeLRU() {
+ return lruCache.pollFirst();
+ }
+
+ public boolean isEmpty() {
+ return lruCache.isEmpty();
+ }
+
+ public int size() {
+ return lruCache.size();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/util/LRUCacheMap.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/LRUCacheMap.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/LRUCacheMap.java
new file mode 100644
index 0000000..6f57e03
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/LRUCacheMap.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.util;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public class LRUCacheMap<K, V> {
+
+ private final LinkedHashMap<K, V> lruCache = new LinkedHashMap<K, V>(16, 0.75F, true);
+
+ public static <K, V> LRUCacheMap<K, V> create() {
+ return new LRUCacheMap<K, V>();
+ }
+
+ public void put(K key, V value) {
+ lruCache.put(key, value);
+ }
+
+ public V get(K key) {
+ return lruCache.get(key);
+ }
+
+ public V getLRU() {
+ for (Map.Entry<K, V> entry : lruCache.entrySet()) {
+ return lruCache.get(entry.getKey());
+ }
+
+ return null;
+ }
+
+ public V remove(K key) {
+ return lruCache.remove(key);
+ }
+
+ public V removeLRU() {
+ for (Map.Entry<K, V> entry : lruCache.entrySet()) {
+ return lruCache.remove(entry.getKey());
+ }
+
+ return null;
+ }
+
+ public boolean containsKey(K key) {
+ return lruCache.containsKey(key);
+ }
+
+ public boolean isEmpty() {
+ return lruCache.isEmpty();
+ }
+
+ public int size() {
+ return lruCache.size();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/EventListener.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/EventListener.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/EventListener.java
new file mode 100644
index 0000000..57ca520
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/EventListener.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.util.event;
+
+public interface EventListener<T> {
+
+ void onEvent(T event);
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/EventNotificationHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/EventNotificationHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/EventNotificationHandler.java
new file mode 100644
index 0000000..a9b3a86
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/EventNotificationHandler.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.util.event;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+
+/**
+ * The event handler manages {@link EventListener} instances and allows to
+ * to publish events to them.
+ */
+public class EventNotificationHandler<T> {
+
+ // Listeners for each event type
+ private final Multimap<Class<? extends T>, EventListener<T>> listeners = HashMultimap.create();
+
+ public void subscribe(EventListener<T> listener, Class<? extends T> eventType) {
+ synchronized (listeners) {
+ listeners.put(eventType, listener);
+ }
+ }
+
+ public void unsubscribe(EventListener<T> listener, Class<? extends T> eventType) {
+ synchronized (listeners) {
+ listeners.remove(eventType, listener);
+ }
+ }
+
+ /**
+ * Publishes the event to all subscribed {@link EventListener} objects.
+ *
+ * @param event The event to publish.
+ */
+ public void publish(T event) {
+ synchronized (listeners) {
+ for (EventListener<T> listener : listeners.get((Class<? extends T>) event.getClass())) {
+ listener.onEvent(event);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/NotificationListener.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/NotificationListener.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/NotificationListener.java
new file mode 100644
index 0000000..6930995
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/event/NotificationListener.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.util.event;
+
+public interface NotificationListener {
+
+ void onNotification();
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/IOReadableWritableSerializer.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/IOReadableWritableSerializer.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/IOReadableWritableSerializer.scala
index 3a906b1..54f26a9 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/IOReadableWritableSerializer.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/IOReadableWritableSerializer.scala
@@ -20,7 +20,8 @@ package org.apache.flink.runtime.akka.serialization
import akka.serialization.JSerializer
import org.apache.flink.core.io.IOReadableWritable
-import org.apache.flink.runtime.io.network.serialization.{DataInputDeserializer, DataOutputSerializer}
+import org.apache.flink.runtime.util.{DataOutputSerializer, DataInputDeserializer}
+
import org.apache.flink.util.InstantiationUtil
class IOReadableWritableSerializer extends JSerializer {
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/WritableSerializer.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/WritableSerializer.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/WritableSerializer.scala
index be1ef17..25f16bd 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/WritableSerializer.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/WritableSerializer.scala
@@ -19,8 +19,7 @@
package org.apache.flink.runtime.akka.serialization
import akka.serialization.JSerializer
-import org.apache.flink.runtime.io.network.serialization.{DataInputDeserializer,
-DataOutputSerializer}
+import org.apache.flink.runtime.util.{DataInputDeserializer, DataOutputSerializer}
import org.apache.flink.util.InstantiationUtil
import org.apache.hadoop.io.Writable
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index 97b3b8c..3dfd70f 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -29,7 +29,6 @@ import org.apache.flink.configuration.{ConfigConstants, GlobalConfiguration, Con
import org.apache.flink.core.io.InputSplitAssigner
import org.apache.flink.runtime.blob.BlobServer
import org.apache.flink.runtime.executiongraph.{Execution, ExecutionJobVertex, ExecutionGraph}
-import org.apache.flink.runtime.io.network.ConnectionInfoLookupResponse
import org.apache.flink.runtime.messages.ArchiveMessages.ArchiveExecutionGraph
import org.apache.flink.runtime.messages.ExecutionGraphMessages.JobStatusChanged
import org.apache.flink.runtime.taskmanager.TaskManager
@@ -361,18 +360,15 @@ Actor with ActorLogMessages with ActorLogging {
}
}
- case LookupConnectionInformation(connectionInformation, jobID, sourceChannelID) => {
- currentJobs.get(jobID) match {
+ case ScheduleOrUpdateConsumers(jobId, executionId, partitionIndex) => {
+ currentJobs.get(jobId) match {
case Some((executionGraph, _)) =>
- val originalSender = sender
- Future {
- originalSender ! ConnectionInformation(
- executionGraph.lookupConnectionInfoAndDeployReceivers
- (connectionInformation, sourceChannelID))
- }
+ sender ! ConsumerNotificationResult(executionGraph
+ .scheduleOrUpdateConsumers(executionId, partitionIndex))
case None =>
- log.error(s"Cannot find execution graph for job ID ${jobID}.")
- sender ! ConnectionInformation(ConnectionInfoLookupResponse.createReceiverNotFound())
+ log.error(s"Cannot find execution graph for job ID ${jobId}.")
+ sender ! ConsumerNotificationResult(false, Some(
+ new IllegalStateException("Cannot find execution graph for job ID " + jobId)))
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
index 3637614..fbacbd2 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobmanagerMessages.scala
@@ -20,10 +20,8 @@ package org.apache.flink.runtime.messages
import org.apache.flink.runtime.accumulators.AccumulatorEvent
import org.apache.flink.runtime.executiongraph.{ExecutionAttemptID, ExecutionGraph}
-import org.apache.flink.runtime.instance.{Instance, InstanceConnectionInfo}
-import org.apache.flink.runtime.io.network.ConnectionInfoLookupResponse
-import org.apache.flink.runtime.io.network.channels.ChannelID
-import org.apache.flink.runtime.jobgraph.{JobStatus, JobVertexID, JobID, JobGraph}
+import org.apache.flink.runtime.instance.Instance
+import org.apache.flink.runtime.jobgraph.{JobGraph, JobID, JobStatus, JobVertexID}
import org.apache.flink.runtime.taskmanager.TaskExecutionState
/**
@@ -75,25 +73,25 @@ object JobManagerMessages {
ExecutionAttemptID)
/**
- * Looks up the connection information of a task being the source of a channel specified by
- * [[sourceChannelID]]. The caller denotes the instance information of the task requesting the
- * lookup information. The connection information is sent back to the sender as a
- * [[ConnectionInformation]] message.
+ * Notifies the [[org.apache.flink.runtime.jobmanager.JobManager]] about available data for a
+ * produced partition.
+ * <p>
+ * There is a call to this method for each
+ * [[org.apache.flink.runtime.executiongraph.ExecutionVertex]] instance once per produced
+ * [[org.apache.flink.runtime.io.network.partition.IntermediateResultPartition]] instance,
+ * either when first producing data (for pipelined executions) or when all data has been produced
+ * (for staged executions).
+ * <p>
+ * The [[org.apache.flink.runtime.jobmanager.JobManager]] then can decide when to schedule the
+ * partition consumers of the given session.
*
- * @param caller instance on which the task requesting the connection information runs
- * @param jobID
- * @param sourceChannelID denoting the channel whose producer shall be found
+ * @see [[org.apache.flink.runtime.io.network.partition.IntermediateResultPartition]]
*/
- case class LookupConnectionInformation(caller: InstanceConnectionInfo, jobID: JobID,
- sourceChannelID: ChannelID)
+ case class ScheduleOrUpdateConsumers(jobId: JobID,
+ executionId: ExecutionAttemptID,
+ partitionIndex: Int)
- /**
- * Contains the connection lookup information of a lookup request triggered by
- * [[LookupConnectionInformation]].
- *
- * @param response
- */
- case class ConnectionInformation(response: ConnectionInfoLookupResponse)
+ case class ConsumerNotificationResult(success: Boolean, error: Option[Throwable] = None)
/**
* Reports the accumulator results of the individual tasks to the job manager.
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
index f69b629..d5e7e9b 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
@@ -19,13 +19,13 @@
package org.apache.flink.runtime.messages
import org.apache.flink.core.io.InputSplit
-import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor
+import org.apache.flink.runtime.deployment.{PartitionInfo, TaskDeploymentDescriptor}
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
-import org.apache.flink.runtime.instance.InstanceID
+import org.apache.flink.runtime.instance.InstanceID
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID
object TaskManagerMessages {
-
/**
* Cancels the task associated with [[attemptID]]. The result is sent back to the sender as a
* [[TaskOperationResult]] message.
@@ -51,13 +51,28 @@ object TaskManagerMessages {
case class NextInputSplit(inputSplit: InputSplit)
/**
- * Unregisters a task identified by [[executionID]] from the task manager.
+ * Unregisters the task identified by [[executionID]] from the task manager.
*
* @param executionID
*/
case class UnregisterTask(executionID: ExecutionAttemptID)
/**
+ * Updates the reader identified by [[resultId]] of the task identified by
+ * [[executionId]] from the task manager.
+ */
+ case class UpdateTask(executionId: ExecutionAttemptID,
+ resultId: IntermediateDataSetID,
+ partitionInfo: PartitionInfo)
+
+ /**
+ * Fails all intermediate result partitions identified by [[executionID]] from the task manager.
+ *
+ * @param executionID
+ */
+ case class FailIntermediateResultPartitions(executionID: ExecutionAttemptID)
+
+ /**
* Reports whether a task manager operation has been successful or not. This message will be
* sent to the sender as a response to [[SubmitTask]] and [[CancelTask]].
*
@@ -115,7 +130,7 @@ object TaskManagerMessages {
*/
case class FailTask(executionID: ExecutionAttemptID, cause: Throwable)
- // --------------------------------------------------------------------------
+ // --------------------------------------------------------------------------
// Utility methods to allow simpler case object access from Java
// --------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
index 67053b2..59174f6 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/LocalFlinkMiniCluster.scala
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.io.FileOutputFormat
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.client.JobClient
+import org.apache.flink.runtime.io.network.netty.NettyConfig
import org.apache.flink.runtime.jobmanager.JobManager
import org.apache.flink.runtime.taskmanager.TaskManager
import org.apache.flink.runtime.util.EnvironmentInformation
@@ -171,6 +172,11 @@ class LocalFlinkMiniCluster(userConfiguration: Configuration, singleActorSystem:
config.setInteger(ConfigConstants.LOCAL_INSTANCE_MANAGER_NUMBER_TASK_MANAGER, 1)
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, ConfigConstants
.DEFAULT_TASK_MANAGER_NUM_TASK_SLOTS)
+
+ // Reduce number of threads for local execution
+ config.setInteger(NettyConfig.NUM_THREADS_CLIENT, 1)
+ config.setInteger(NettyConfig.NUM_THREADS_SERVER, 2)
+
config
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkConnectionConfiguration.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkConnectionConfiguration.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkConnectionConfiguration.scala
deleted file mode 100644
index a76ae2a..0000000
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkConnectionConfiguration.scala
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.taskmanager
-
-sealed trait NetworkConnectionConfiguration{
- def numBuffers: Int
- def bufferSize: Int
-}
-
-case class LocalNetworkConfiguration(numBuffers: Int,
- bufferSize: Int) extends NetworkConnectionConfiguration
-
-case class ClusterNetworkConfiguration(numBuffers: Int, bufferSize: Int, numInThreads: Int,
- numOutThreads: Int, lowWaterMark: Int,
- highWaterMark: Int) extends NetworkConnectionConfiguration
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
new file mode 100644
index 0000000..aeda6c4
--- /dev/null
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.taskmanager
+
+import org.apache.flink.runtime.io.network.netty.NettyConfig
+
+case class NetworkEnvironmentConfiguration(numNetworkBuffers: Int,
+ networkBufferSize: Int,
+ nettyConfig: Option[NettyConfig] = None)
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 4676ae9..620a436 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -18,75 +18,82 @@
package org.apache.flink.runtime.taskmanager
-import java.io.{IOException, File}
-import java.lang.management.{GarbageCollectorMXBean, MemoryMXBean, ManagementFactory}
+import java.io.{File, IOException}
import java.net.{InetAddress, InetSocketAddress}
import java.util
import java.util.concurrent.{FutureTask, TimeUnit}
-import scala.collection.JavaConverters._
+import management.{GarbageCollectorMXBean, ManagementFactory, MemoryMXBean}
+
import akka.actor._
import akka.pattern.ask
import org.apache.flink.api.common.cache.DistributedCache
-import org.apache.flink.configuration.{GlobalConfiguration, ConfigConstants, Configuration}
+import org.apache.flink.configuration.{ConfigConstants, Configuration, GlobalConfiguration}
import org.apache.flink.core.fs.Path
import org.apache.flink.runtime.ActorLogMessages
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.blob.BlobCache
import org.apache.flink.runtime.broadcast.BroadcastVariableManager
-import org.apache.flink.runtime.execution.{CancelTaskException, ExecutionState, RuntimeEnvironment}
+import org.apache.flink.runtime.deployment.{PartitionInfo, TaskDeploymentDescriptor}
import org.apache.flink.runtime.execution.librarycache.{BlobLibraryCacheManager, FallbackLibraryCacheManager, LibraryCacheManager}
+import org.apache.flink.runtime.execution.{CancelTaskException, ExecutionState, RuntimeEnvironment}
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
import org.apache.flink.runtime.filecache.FileCache
-import org.apache.flink.runtime.instance.{InstanceConnectionInfo, HardwareDescription, InstanceID}
-import org.apache.flink.runtime.io.disk.iomanager.{IOManagerAsync}
-import org.apache.flink.runtime.io.network.netty.NettyConnectionManager
-import org.apache.flink.runtime.io.network.{NetworkConnectionManager, LocalConnectionManager, ChannelManager}
-import org.apache.flink.runtime.jobgraph.JobID
+import org.apache.flink.runtime.instance.{HardwareDescription, InstanceConnectionInfo, InstanceID}
+import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync
+import org.apache.flink.runtime.io.network.NetworkEnvironment
+import org.apache.flink.runtime.io.network.netty.NettyConfig
+import org.apache.flink.runtime.jobgraph.{IntermediateDataSetID, JobID}
import org.apache.flink.runtime.jobmanager.JobManager
import org.apache.flink.runtime.memorymanager.DefaultMemoryManager
import org.apache.flink.runtime.messages.JobManagerMessages.UpdateTaskExecutionState
-import org.apache.flink.runtime.messages.RegistrationMessages.{RegisterTaskManager, AcknowledgeRegistration}
+import org.apache.flink.runtime.messages.RegistrationMessages.{AcknowledgeRegistration, RegisterTaskManager}
import org.apache.flink.runtime.messages.TaskManagerMessages._
-import org.apache.flink.runtime.messages.TaskManagerProfilerMessages.{UnmonitorTask, MonitorTask, RegisterProfilingListener}
+import org.apache.flink.runtime.messages.TaskManagerProfilerMessages.{MonitorTask, RegisterProfilingListener, UnmonitorTask}
import org.apache.flink.runtime.net.NetUtils
import org.apache.flink.runtime.profiling.ProfilingUtils
import org.apache.flink.runtime.util.EnvironmentInformation
import org.apache.flink.util.ExceptionUtils
import org.slf4j.LoggerFactory
+
import scala.concurrent.Future
import scala.concurrent.duration._
-import scala.util.Failure
-import scala.util.Success
-import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor
+import scala.util.{Failure, Success}
/**
- *
- *
+ *
+ *
* The TaskManager has the following phases:
- *
- * - Waiting to be registered with its JobManager. In that phase, it periodically sends
- * [[RegisterAtJobManager]] messages to itself, which trigger the sending of
- * a [[RegisterTaskManager]] message to the JobManager.
- *
- * - Upon successful registration, the JobManager replies with an [[AcknowledgeRegistration]]
- * message. This stops the registration messages and initializes all fields
- * that require the JobManager's actor reference
- *
- * - ...
+ *
+ * - Waiting to be registered with its JobManager. In that phase, it periodically sends
+ * [[RegisterAtJobManager]] messages to itself, which trigger the sending of
+ * a [[RegisterTaskManager]] message to the JobManager.
+ *
+ * - Upon successful registration, the JobManager replies with an [[AcknowledgeRegistration]]
+ * message. This stops the registration messages and initializes all fields
+ * that require the JobManager's actor reference
+ *
+ * - ...
*/
class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkkaURL: String,
val taskManagerConfig: TaskManagerConfiguration,
- val networkConnectionConfig: NetworkConnectionConfiguration)
+ val networkConfig: NetworkEnvironmentConfiguration)
extends Actor with ActorLogMessages with ActorLogging {
import context._
import taskManagerConfig.{timeout => tmTimeout, _}
+
+import scala.collection.JavaConverters._
+
implicit val timeout = tmTimeout
log.info(s"Starting task manager at ${self.path}.")
+ val REGISTRATION_DELAY = 0 seconds
+ val REGISTRATION_INTERVAL = 10 seconds
+ val MAX_REGISTRATION_ATTEMPTS = 10
+ val HEARTBEAT_INTERVAL = 5000 millisecond
+
TaskManager.checkTempDirs(tmpDirPaths)
-
val ioManager = new IOManagerAsync(tmpDirPaths)
val memoryManager = new DefaultMemoryManager(memorySize, numberOfSlots, pageSize)
val bcVarManager = new BroadcastVariableManager();
@@ -98,13 +105,13 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
val waitForRegistration = scala.collection.mutable.Set[ActorRef]();
val profiler = profilingInterval match {
- case Some(interval) =>
- Some(TaskManager.startProfiler(self.path.toSerializationFormat, interval))
+ case Some(interval) => Some(TaskManager.startProfiler(self.path.toSerializationFormat,
+ interval))
case None => None
}
- var libraryCacheManager: LibraryCacheManager = _
- var channelManager: Option[ChannelManager] = None
+ var libraryCacheManager: LibraryCacheManager = null
+ var networkEnvironment: Option[NetworkEnvironment] = None
var registrationScheduler: Option[Cancellable] = None
var registrationAttempts: Int = 0
var registered: Boolean = false
@@ -133,10 +140,10 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
_.cancel()
}
- channelManager foreach {
- channelManager =>
+ networkEnvironment foreach {
+ ne =>
try {
- channelManager.shutdown()
+ ne.shutdown()
} catch {
case t: Throwable => log.error(t, "ChannelManager did not shutdown properly.")
}
@@ -160,11 +167,10 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
registrationAttempts = 0
import context.dispatcher
registrationScheduler = Some(context.system.scheduler.schedule(
- TaskManager.REGISTRATION_DELAY, TaskManager.REGISTRATION_INTERVAL,
- self, RegisterAtJobManager))
+ TaskManager.REGISTRATION_DELAY, TaskManager.REGISTRATION_INTERVAL,
+ self, RegisterAtJobManager))
}
-
override def receiveWithLogMessages: Receive = {
case RegisterAtJobManager => {
registrationAttempts += 1
@@ -195,13 +201,14 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
context.watch(currentJobManager)
log.info(s"TaskManager successfully registered at JobManager ${
- currentJobManager.path.toString }.")
+ currentJobManager.path.toString
+ }.")
- setupChannelManager()
+ setupNetworkEnvironment()
setupLibraryCacheManager(blobPort)
heartbeatScheduler = Some(context.system.scheduler.schedule(
- TaskManager.HEARTBEAT_INTERVAL, TaskManager.HEARTBEAT_INTERVAL, self, SendHeartbeat))
+ TaskManager.HEARTBEAT_INTERVAL, TaskManager.HEARTBEAT_INTERVAL, self, SendHeartbeat))
profiler foreach {
_.tell(RegisterProfilingListener, JobManager.getProfiler(currentJobManager))
@@ -214,11 +221,15 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
waitForRegistration.clear()
}
}
-
+
case SubmitTask(tdd) => {
submitTask(tdd)
}
+ case UpdateTask(executionId, resultId, partitionInfo) => {
+ updateTask(executionId, resultId, partitionInfo)
+ }
+
case CancelTask(executionID) => {
runningTasks.get(executionID) match {
case Some(task) =>
@@ -228,7 +239,7 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
sender ! new TaskOperationResult(executionID, true)
case None =>
sender ! new TaskOperationResult(executionID, false,
- "No task with that execution ID was found.")
+ "No task with that execution ID was found.")
}
}
@@ -245,17 +256,17 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
}
case NotifyWhenRegisteredAtJobManager => {
- if (registered) {
- sender ! RegisteredAtJobManager
- } else {
- waitForRegistration += sender
+ if (registered) {
+ sender ! RegisteredAtJobManager
+ } else {
+ waitForRegistration += sender
}
}
case FailTask(executionID, cause) => {
runningTasks.get(executionID) match {
case Some(task) =>
- Future{
+ Future {
task.failExternally(cause)
}
case None =>
@@ -264,8 +275,8 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
case Terminated(jobManager) => {
log.info(s"Job manager ${jobManager.path} is no longer reachable. "
- + "Cancelling all tasks and trying to reregister.")
-
+ + "Cancelling all tasks and trying to reregister.")
+
cancelAndClearEverything(new Throwable("Lost connection to JobManager"))
tryJobManagerRegistration()
}
@@ -280,7 +291,7 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
futureResponse.mapTo[Boolean].onComplete {
case Success(result) =>
- if(!result){
+ if (!result) {
self ! FailTask(executionID, new IllegalStateException("Task has been disposed on " +
"JobManager."))
}
@@ -297,12 +308,12 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
}
}
- private def submitTask(tdd: TaskDeploymentDescriptor) : Unit = {
+ private def submitTask(tdd: TaskDeploymentDescriptor): Unit = {
val jobID = tdd.getJobID
val vertexID = tdd.getVertexID
val executionID = tdd.getExecutionId
val taskIndex = tdd.getIndexInSubtaskGroup
- val numSubtasks = tdd.getCurrentNumberOfSubtasks
+ val numSubtasks = tdd.getNumberOfSubtasks
var startRegisteringTask = 0L
var task: Task = null
@@ -314,7 +325,8 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
if (log.isDebugEnabled) {
log.debug(s"Register task ${executionID} took ${
- (System.currentTimeMillis() - startRegisteringTask)/1000.0}s")
+ (System.currentTimeMillis() - startRegisteringTask) / 1000.0
+ }s")
}
val userCodeClassLoader = libraryCacheManager.getClassLoader(jobID)
@@ -328,99 +340,127 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
runningTasks.put(executionID, task) match {
case Some(_) => throw new RuntimeException(
- s"TaskManager contains already a task with executionID ${executionID}.")
+ s"TaskManager contains already a task with executionID ${executionID}.")
case None =>
}
val splitProvider = new TaskInputSplitProvider(currentJobManager, jobID, vertexID,
- executionID, timeout)
+ executionID, timeout)
- val env = new RuntimeEnvironment(task, tdd, userCodeClassLoader, memoryManager,
- ioManager, splitProvider, currentJobManager, bcVarManager)
+ val env = new RuntimeEnvironment(currentJobManager, task, tdd, userCodeClassLoader,
+ memoryManager, ioManager, splitProvider, bcVarManager, networkEnvironment.get)
task.setEnvironment(env)
// register the task with the network stack and profilers
- channelManager match {
- case Some(cm) => cm.register(task)
- case None => throw new RuntimeException("ChannelManager has not been properly " +
- "instantiated.")
- }
+ networkEnvironment match {
+ case Some(ne) => ne.registerTask(task)
+ case None => throw new RuntimeException(
+ "Network environment has not been properly instantiated.")
+ }
- val jobConfig = tdd.getJobConfiguration
+ val jobConfig = tdd.getJobConfiguration
- if (jobConfig.getBoolean(ProfilingUtils.PROFILE_JOB_KEY, true)) {
- profiler match {
- case Some(profiler) => profiler ! MonitorTask(task)
- case None => log.info("There is no profiling enabled for the task manager.")
- }
+ if (jobConfig.getBoolean(ProfilingUtils.PROFILE_JOB_KEY, true)) {
+ profiler match {
+ case Some(profiler) => profiler ! MonitorTask(task)
+ case None => log.info("There is no profiling enabled for the task manager.")
}
+ }
- val cpTasks = new util.HashMap[String, FutureTask[Path]]()
+ val cpTasks = new util.HashMap[String, FutureTask[Path]]()
- for (entry <- DistributedCache.readFileInfoFromConfig(tdd.getJobConfiguration).asScala) {
- val cp = fileCache.createTmpFile(entry.getKey, entry.getValue, jobID)
- cpTasks.put(entry.getKey, cp)
- }
- env.addCopyTasksForCacheFile(cpTasks)
+ for (entry <- DistributedCache.readFileInfoFromConfig(tdd.getJobConfiguration).asScala) {
+ val cp = fileCache.createTmpFile(entry.getKey, entry.getValue, jobID)
+ cpTasks.put(entry.getKey, cp)
+ }
+ env.addCopyTasksForCacheFile(cpTasks)
+
+ if (!task.startExecution()) {
+ throw new RuntimeException("Cannot start task. Task was canceled or failed.")
+ }
- if (!task.startExecution()) {
- throw new RuntimeException("Cannot start task. Task was canceled or failed.")
+ sender ! TaskOperationResult(executionID, true)
+ } catch {
+ case t: Throwable =>
+ val message = if (t.isInstanceOf[CancelTaskException]) {
+ "Task was canceled"
+ } else {
+ log.error(t, s"Could not instantiate task with execution ID ${executionID}.")
+ ExceptionUtils.stringifyException(t)
}
- sender ! TaskOperationResult(executionID, true)
- } catch {
- case t: Throwable =>
- val message = if(t.isInstanceOf[CancelTaskException]){
- "Task was canceled"
- }else{
- log.error(t, s"Could not instantiate task with execution ID ${executionID}.")
- ExceptionUtils.stringifyException(t)
+ try {
+ if (task != null) {
+ task.failExternally(t)
+ removeAllTaskResources(task)
}
- try {
- if (task != null) {
- task.failExternally(t)
- removeAllTaskResources(task)
- }
+ libraryCacheManager.unregisterTask(jobID, executionID)
+ } catch {
+ case t: Throwable => log.error("Error during cleanup of task deployment.", t)
+ }
- libraryCacheManager.unregisterTask(jobID, executionID)
- } catch {
- case t: Throwable => log.error("Error during cleanup of task deployment.", t)
- }
+ sender ! new TaskOperationResult(executionID, false, message)
+ }
+ }
- sender ! new TaskOperationResult(executionID, false, message)
- }
+ private def updateTask(executionId: ExecutionAttemptID, resultId: IntermediateDataSetID,
+ partitionInfo: PartitionInfo): Unit = {
+
+ partitionInfo.getProducerLocation match {
+ case PartitionInfo.PartitionLocation.UNKNOWN =>
+ sender ! TaskOperationResult(executionId, false,
+ "Tried to update task with UNKNOWN channel.")
+
+ case _ =>
+ runningTasks.get(executionId) match {
+ case Some(task) =>
+ Option(task.getEnvironment.getReaderById(resultId)) match {
+ case Some(reader) =>
+ Future {
+ try {
+ reader.updateInputChannel(partitionInfo)
+ } catch {
+ case t: Throwable =>
+ log.error("Task update failure: {} Trying to cancel task.", t.getMessage)
+
+ try {
+ task.cancelExecution()
+ } catch {
+ case t: Throwable =>
+ log.error("Failed canceling task with execution ID {} after task" +
+ "update failure: {}.", executionId, t.getMessage)
+ }
+ }
+ }
+ sender ! TaskOperationResult(executionId, true)
+ case None => sender ! TaskOperationResult(executionId, false, "No reader with ID " +
+ resultId + " was found.");
+ }
+
+ case None => sender ! TaskOperationResult(executionId, false, "No task with execution" +
+ "ID " + executionId + " was found.");
+ }
+ }
}
-
- private def setupChannelManager(): Unit = {
- //shutdown existing channel manager
- channelManager foreach {
- cm =>
+
+ def setupNetworkEnvironment(): Unit = {
+ //shutdown existing network environment
+ networkEnvironment foreach {
+ ne =>
try {
- cm.shutdown()
+ ne.shutdown()
} catch {
- case t: Throwable => log.error(t, "ChannelManager did not shutdown properly.")
+ case t: Throwable => log.error(t, "Network environment did not shutdown properly.")
}
}
try {
- import networkConnectionConfig._
-
- val connectionManager: NetworkConnectionManager = networkConnectionConfig match {
- case _: LocalNetworkConfiguration => new LocalConnectionManager
- case ClusterNetworkConfiguration(numBuffers, bufferSize, numInThreads,
- numOutThreads, lowWaterMark, highWaterMark) =>
- new NettyConnectionManager(connectionInfo.address(),
- connectionInfo.dataPort(), bufferSize, numInThreads, numOutThreads, lowWaterMark,
- highWaterMark)
- }
-
- channelManager = Some(new ChannelManager(currentJobManager, connectionInfo, numBuffers,
- bufferSize, connectionManager, timeout))
+ networkEnvironment = Some(new NetworkEnvironment(currentJobManager, timeout, networkConfig))
} catch {
case ioe: IOException =>
- log.error(ioe, "Failed to instantiate ChannelManager.")
+ log.error(ioe, "Failed to instantiate network environment.")
throw new RuntimeException("Failed to instantiate ChannelManager.", ioe)
}
}
@@ -436,7 +476,7 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
}
libraryCacheManager = null
}
-
+
if (blobPort > 0) {
val address = new InetSocketAddress(currentJobManager.path.address.host.getOrElse
("localhost"), blobPort)
@@ -466,27 +506,27 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
removeAllTaskResources(task)
libraryCacheManager.unregisterTask(task.getJobID, executionID)
case None =>
- if(log.isDebugEnabled){
+ if (log.isDebugEnabled) {
log.debug(s"Cannot find task with ID ${executionID} to unregister.")
}
}
}
private def removeAllTaskResources(task: Task): Unit = {
- if(task.getEnvironment != null) {
+ if (task.getEnvironment != null) {
try {
for (entry <- DistributedCache.readFileInfoFromConfig(task.getEnvironment
.getJobConfiguration).asScala) {
fileCache.deleteTmpFile(entry.getKey, entry.getValue, task.getJobID)
}
- }catch{
+ } catch {
case t: Throwable => log.error("Error cleaning up local files from the distributed cache" +
".", t)
}
}
- channelManager foreach {
- _.unregister(task.getExecutionId, task)
+ networkEnvironment foreach {
+ _.unregisterTask(task)
}
profiler foreach {
@@ -495,12 +535,12 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
task.unregisterMemoryManager(memoryManager)
}
-
+
private def logMemoryStats(): Unit = {
if (log.isDebugEnabled) {
val memoryMXBean = ManagementFactory.getMemoryMXBean()
val gcMXBeans = ManagementFactory.getGarbageCollectorMXBeans().asScala
-
+
log.debug(TaskManager.getMemoryUsageStatsAsString(memoryMXBean))
log.debug(TaskManager.getGarbageCollectorStatsAsString(gcMXBeans))
}
@@ -518,13 +558,12 @@ object TaskManager {
val TASK_MANAGER_NAME = "taskmanager"
val PROFILER_NAME = "profiler"
-
+
val REGISTRATION_DELAY = 0 seconds
val REGISTRATION_INTERVAL = 10 seconds
val MAX_REGISTRATION_ATTEMPTS = 10
val HEARTBEAT_INTERVAL = 5000 millisecond
-
def main(args: Array[String]): Unit = {
val (hostname, port, configuration) = parseArgs(args)
@@ -580,16 +619,16 @@ object TaskManager {
localExecution: Boolean = false): (ActorSystem, ActorRef) = {
implicit val actorSystem = AkkaUtils.createActorSystem(hostname, port, configuration)
- val (connectionInfo, jobManagerURL, taskManagerConfig, networkConnectionConfiguration) =
+ val (connectionInfo, jobManagerURL, taskManagerConfig, networkConfig) =
parseConfiguration(hostname, configuration, localExecution)
(actorSystem, startActor(connectionInfo, jobManagerURL, taskManagerConfig,
- networkConnectionConfiguration))
+ networkConfig))
}
def parseConfiguration(hostname: String, configuration: Configuration,
localExecution: Boolean = false):
- (InstanceConnectionInfo, String, TaskManagerConfiguration, NetworkConnectionConfiguration) = {
+ (InstanceConnectionInfo, String, TaskManagerConfiguration, NetworkEnvironmentConfiguration) = {
val dataport = configuration.getInteger(ConfigConstants.TASK_MANAGER_DATA_PORT_KEY,
ConfigConstants.DEFAULT_TASK_MANAGER_DATA_PORT) match {
case 0 => NetUtils.getAvailablePort
@@ -624,32 +663,19 @@ object TaskManager {
val tmpDirs = configuration.getString(ConfigConstants.TASK_MANAGER_TMP_DIR_KEY,
ConfigConstants.DEFAULT_TASK_MANAGER_TMP_PATH).split(",|" + File.pathSeparator)
- val numBuffers = configuration.getInteger(ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY,
+ val numNetworkBuffers = configuration.getInteger(
+ ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY,
ConfigConstants.DEFAULT_TASK_MANAGER_NETWORK_NUM_BUFFERS)
- val bufferSize = configuration.getInteger(ConfigConstants.TASK_MANAGER_NETWORK_BUFFER_SIZE_KEY,
- ConfigConstants.DEFAULT_TASK_MANAGER_NETWORK_BUFFER_SIZE)
- val networkConnectionConfiguration = if(localExecution){
- LocalNetworkConfiguration(numBuffers, bufferSize)
- } else {
- val numInThreads = configuration.getInteger(
- ConfigConstants.TASK_MANAGER_NET_NUM_IN_THREADS_KEY,
- ConfigConstants.DEFAULT_TASK_MANAGER_NET_NUM_IN_THREADS)
- val numOutThreads = configuration.getInteger(ConfigConstants
- .TASK_MANAGER_NET_NUM_OUT_THREADS_KEY,
- ConfigConstants.DEFAULT_TASK_MANAGER_NET_NUM_OUT_THREADS)
- val lowWaterMark = configuration.getInteger(ConfigConstants
- .TASK_MANAGER_NET_NETTY_LOW_WATER_MARK,
- ConfigConstants.DEFAULT_TASK_MANAGER_NET_NETTY_LOW_WATER_MARK)
- val highWaterMark = configuration.getInteger(ConfigConstants
- .TASK_MANAGER_NET_NETTY_HIGH_WATER_MARK,
- ConfigConstants.DEFAULT_TASK_MANAGER_NET_NETTY_HIGH_WATER_MARK)
-
- ClusterNetworkConfiguration(numBuffers, bufferSize, numInThreads, numOutThreads,
- lowWaterMark, highWaterMark)
+ val nettyConfig = localExecution match {
+ case true => None
+ case false => Some(new NettyConfig(
+ connectionInfo.address(), connectionInfo.dataPort(), pageSize, configuration))
}
- val networkBufferMem = if(localExecution) 0 else numBuffers * bufferSize;
+ val networkConfig = NetworkEnvironmentConfiguration(numNetworkBuffers, pageSize, nettyConfig)
+
+ val networkBufferMem = if (localExecution) 0 else numNetworkBuffers * pageSize;
val configuredMemory: Long = configuration.getInteger(ConfigConstants
.TASK_MANAGER_MEMORY_SIZE_KEY, -1)
@@ -659,11 +685,10 @@ object TaskManager {
} else {
val fraction = configuration.getFloat(ConfigConstants.TASK_MANAGER_MEMORY_FRACTION_KEY,
ConfigConstants.DEFAULT_MEMORY_MANAGER_MEMORY_FRACTION)
- ((EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag - networkBufferMem ) * fraction)
+ ((EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag - networkBufferMem) * fraction)
.toLong
}
-
val memoryLoggingIntervalMs = configuration.getBoolean(ConfigConstants
.TASK_MANAGER_DEBUG_MEMORY_USAGE_START_LOG_THREAD,
ConfigConstants.DEFAULT_TASK_MANAGER_DEBUG_MEMORY_USAGE_START_LOG_THREAD) match {
@@ -691,15 +716,15 @@ object TaskManager {
val taskManagerConfig = TaskManagerConfiguration(numberOfSlots, memorySize, pageSize,
tmpDirs, cleanupInterval, memoryLoggingIntervalMs, profilingInterval, timeout)
- (connectionInfo, jobManagerURL, taskManagerConfig, networkConnectionConfiguration)
+ (connectionInfo, jobManagerURL, taskManagerConfig, networkConfig)
}
def startActor(connectionInfo: InstanceConnectionInfo, jobManagerURL: String,
taskManagerConfig: TaskManagerConfiguration,
- networkConnectionConfiguration: NetworkConnectionConfiguration)
+ networkConfig: NetworkEnvironmentConfiguration)
(implicit actorSystem: ActorSystem): ActorRef = {
startActor(Props(new TaskManager(connectionInfo, jobManagerURL, taskManagerConfig,
- networkConnectionConfiguration)))
+ networkConfig)))
}
def startActor(props: Props)(implicit actorSystem: ActorSystem): ActorRef = {
@@ -756,9 +781,8 @@ object TaskManager {
s"NON HEAP: $nonHeapUsed/$nonHeapCommitted/$nonHeapMax MB (used/committed/max)]"
}
- private def getGarbageCollectorStatsAsString(gcMXBeans: Iterable[GarbageCollectorMXBean])
- : String =
- {
+ private def getGarbageCollectorStatsAsString(gcMXBeans: Iterable[GarbageCollectorMXBean])
+ : String = {
val beans = gcMXBeans map {
bean =>
s"[${bean.getName}, GC TIME (ms): ${bean.getCollectionTime}, " +
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptorTest.java
deleted file mode 100644
index 811bfcf..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptorTest.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.deployment;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.fail;
-
-import java.io.IOException;
-
-import org.apache.flink.runtime.io.network.channels.ChannelID;
-import org.apache.flink.runtime.testutils.CommonTestUtils;
-import org.apache.flink.util.StringUtils;
-import org.junit.Test;
-
-
-public class ChannelDeploymentDescriptorTest {
-
- /**
- * Tests the constructor of the {@link ChannelDeploymentDescriptor} class with valid arguments.
- */
- @Test
- public void testConstructorWithValidArguments() {
-
- final ChannelID outputChannelID = new ChannelID();
- final ChannelID inputChannelID = new ChannelID();
-
- final ChannelDeploymentDescriptor cdd = new ChannelDeploymentDescriptor(outputChannelID, inputChannelID);
-
- assertEquals(outputChannelID, cdd.getOutputChannelID());
- assertEquals(inputChannelID, cdd.getInputChannelID());
- }
-
- /**
- * Tests the constructor of the {@link ChannelDeploymentDescriptor} class with invalid arguments.
- */
- @Test
- public void testConstructorWithInvalidArguments() {
-
- final ChannelID channelID = new ChannelID();
-
- boolean firstExceptionCaught = false;
- boolean secondExceptionCaught = false;
-
- try {
-
- new ChannelDeploymentDescriptor(null, channelID);
-
- } catch (IllegalArgumentException e) {
- firstExceptionCaught = true;
- }
-
- try {
-
- new ChannelDeploymentDescriptor(channelID, null);
-
- } catch (IllegalArgumentException e) {
- secondExceptionCaught = true;
- }
-
- if (!firstExceptionCaught) {
- fail("First argument was illegal but not detected");
- }
-
- if (!secondExceptionCaught) {
- fail("Second argument was illegal but not detected");
- }
- }
-
- /**
- * Tests the serialization/deserialization of the {@link ChannelDeploymentDescriptor} class.
- */
- @Test
- public void testSerialization() {
-
- final ChannelID outputChannelID = new ChannelID();
- final ChannelID inputChannelID = new ChannelID();
-
- final ChannelDeploymentDescriptor orig = new ChannelDeploymentDescriptor(outputChannelID, inputChannelID);
-
- ChannelDeploymentDescriptor copy = null;
-
- try {
-
- copy = CommonTestUtils.createCopySerializable(orig);
- } catch (IOException ioe) {
- fail(StringUtils.stringifyException(ioe));
- }
-
- assertFalse(orig.getOutputChannelID() == copy.getOutputChannelID());
- assertFalse(orig.getInputChannelID() == copy.getInputChannelID());
-
- assertEquals(orig.getOutputChannelID(), copy.getOutputChannelID());
- assertEquals(orig.getInputChannelID(), copy.getInputChannelID());
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorTest.java
index 8bd97dd..1a274d0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorTest.java
@@ -48,13 +48,13 @@ public class TaskDeploymentDescriptorTest {
final Configuration jobConfiguration = new Configuration();
final Configuration taskConfiguration = new Configuration();
final Class<? extends AbstractInvokable> invokableClass = RegularPactTask.class;
- final List<GateDeploymentDescriptor> outputGates = new ArrayList<GateDeploymentDescriptor>(0);
- final List<GateDeploymentDescriptor> inputGates = new ArrayList<GateDeploymentDescriptor>(0);
+ final List<PartitionDeploymentDescriptor> producedResults = new ArrayList<PartitionDeploymentDescriptor>(0);
+ final List<PartitionConsumerDeploymentDescriptor> inputGates = new ArrayList<PartitionConsumerDeploymentDescriptor>(0);
final List<BlobKey> requiredJars = new ArrayList<BlobKey>(0);
final TaskDeploymentDescriptor orig = new TaskDeploymentDescriptor(jobID, vertexID, execId, taskName,
indexInSubtaskGroup, currentNumberOfSubtasks, jobConfiguration, taskConfiguration,
- invokableClass.getName(), outputGates, inputGates, requiredJars, 47);
+ invokableClass.getName(), producedResults, inputGates, requiredJars, 47);
final TaskDeploymentDescriptor copy = CommonTestUtils.createCopySerializable(orig);
@@ -68,9 +68,9 @@ public class TaskDeploymentDescriptorTest {
assertEquals(orig.getVertexID(), copy.getVertexID());
assertEquals(orig.getTaskName(), copy.getTaskName());
assertEquals(orig.getIndexInSubtaskGroup(), copy.getIndexInSubtaskGroup());
- assertEquals(orig.getCurrentNumberOfSubtasks(), copy.getCurrentNumberOfSubtasks());
- assertEquals(orig.getOutputGates(), copy.getOutputGates());
- assertEquals(orig.getInputGates(), copy.getInputGates());
+ assertEquals(orig.getNumberOfSubtasks(), copy.getNumberOfSubtasks());
+ assertEquals(orig.getProducedPartitions(), copy.getProducedPartitions());
+ assertEquals(orig.getConsumedPartitions(), copy.getConsumedPartitions());
assertEquals(orig.getRequiredJarFiles(), copy.getRequiredJarFiles());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d908ca19/flink-runtime/src/test/java/org/apache/flink/runtime/event/task/EventNotificationManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/event/task/EventNotificationManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/event/task/EventNotificationManagerTest.java
deleted file mode 100644
index 9ed1986..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/event/task/EventNotificationManagerTest.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.event.task;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-
-import org.junit.Test;
-
-/**
- * This class contains unit tests for the {@link EventNotificationManager}.
- *
- */
-public class EventNotificationManagerTest {
- /**
- * A test implementation of an {@link EventListener}.
- *
- */
- private static class TestEventListener implements EventListener {
-
- /**
- * The event that was last received by this event listener.
- */
- private AbstractTaskEvent receivedEvent = null;
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void eventOccurred(AbstractTaskEvent event) {
-
- this.receivedEvent = event;
- }
-
- /**
- * Returns the event which was last received by this event listener. If no event
- * has been received so far the return value is <code>null</code>.
- *
- * @return the event which was last received, possibly <code>null</code>
- */
- public AbstractTaskEvent getLastReceivedEvent() {
-
- return this.receivedEvent;
- }
- }
-
- /**
- * Tests the publish/subscribe mechanisms implemented in the {@link EventNotificationManager}.
- */
- @Test
- public void testEventNotificationManager() {
-
- final EventNotificationManager evm = new EventNotificationManager();
- final TestEventListener listener = new TestEventListener();
-
- evm.subscribeToEvent(listener, StringTaskEvent.class);
-
- final StringTaskEvent stringTaskEvent1 = new StringTaskEvent("Test 1");
- final StringTaskEvent stringTaskEvent2 = new StringTaskEvent("Test 2");
-
- evm.deliverEvent(stringTaskEvent1);
- evm.deliverEvent(new IntegerTaskEvent(5));
-
- assertNotNull(listener.getLastReceivedEvent());
- StringTaskEvent receivedStringEvent = (StringTaskEvent) listener.getLastReceivedEvent();
- assertEquals(stringTaskEvent1, receivedStringEvent);
-
- evm.unsubscribeFromEvent(listener, StringTaskEvent.class);
-
- evm.deliverEvent(stringTaskEvent2);
-
- assertNotNull(listener.getLastReceivedEvent());
- receivedStringEvent = (StringTaskEvent) listener.getLastReceivedEvent();
- assertEquals(stringTaskEvent1, receivedStringEvent);
- }
-}