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 2018/01/05 18:38:57 UTC
[16/19] flink git commit: [hotfix] [checkpoints] Improve performance
of ByteStreamStateHandle
[hotfix] [checkpoints] Improve performance of ByteStreamStateHandle
The input stream from ByteStreamStateHandle did not overwrite the 'read(byte[], int, int)' method,
meaning that bulk byte reads resulted in many individual byte accesses.
Additionally, this change avoids accessing the data array through an outer class, but instead adds
a reference directly to the input stream class, avoiding one hop per access. That also allows
a more restricted access level on the fields, which may additionally help the jitter in some cases.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/1d38e0b4
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/1d38e0b4
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/1d38e0b4
Branch: refs/heads/master
Commit: 1d38e0b49a1936fef477a7a2a65abdd815f2d695
Parents: 0ed264b
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Dec 6 14:51:22 2017 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Jan 5 19:38:07 2018 +0100
----------------------------------------------------------------------
.../apache/flink/core/fs/FSDataInputStream.java | 2 +-
.../state/memory/ByteStreamStateHandle.java | 38 +++-
.../state/memory/ByteStreamStateHandleTest.java | 177 +++++++++++++++++++
.../StateInitializationContextImplTest.java | 14 +-
4 files changed, 214 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/1d38e0b4/flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java b/flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java
index fa931c6..5fc1de8 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java
@@ -35,7 +35,7 @@ public abstract class FSDataInputStream extends InputStream {
/**
* Seek to the given offset from the start of the file. The next read() will be from that location.
- * Can't seek past the end of the file.
+ * Can't seek past the end of the stream.
*
* @param desired
* the desired offset
http://git-wip-us.apache.org/repos/asf/flink/blob/1d38e0b4/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandle.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandle.java
index 3a43d4f..7de66ef 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandle.java
@@ -34,13 +34,13 @@ public class ByteStreamStateHandle implements StreamStateHandle {
/**
* The state data.
*/
- protected final byte[] data;
+ private final byte[] data;
/**
* A unique name of by which this state handle is identified and compared. Like a filename, all
* {@link ByteStreamStateHandle} with the exact same name must also have the exact same content in data.
*/
- protected final String handleName;
+ private final String handleName;
/**
* Creates a new ByteStreamStateHandle containing the given data.
@@ -52,7 +52,7 @@ public class ByteStreamStateHandle implements StreamStateHandle {
@Override
public FSDataInputStream openInputStream() throws IOException {
- return new ByteStateHandleInputStream();
+ return new ByteStateHandleInputStream(data);
}
public byte[] getData() {
@@ -81,7 +81,6 @@ public class ByteStreamStateHandle implements StreamStateHandle {
return false;
}
-
ByteStreamStateHandle that = (ByteStreamStateHandle) o;
return handleName.equals(that.handleName);
}
@@ -102,18 +101,22 @@ public class ByteStreamStateHandle implements StreamStateHandle {
/**
* An input stream view on a byte array.
*/
- private final class ByteStateHandleInputStream extends FSDataInputStream {
+ private static final class ByteStateHandleInputStream extends FSDataInputStream {
+ private final byte[] data;
private int index;
- public ByteStateHandleInputStream() {
- this.index = 0;
+ public ByteStateHandleInputStream(byte[] data) {
+ this.data = data;
}
@Override
public void seek(long desired) throws IOException {
- Preconditions.checkArgument(desired >= 0 && desired < Integer.MAX_VALUE);
- index = (int) desired;
+ if (desired >= 0 && desired <= data.length) {
+ index = (int) desired;
+ } else {
+ throw new IOException("position out of bounds");
+ }
}
@Override
@@ -125,5 +128,22 @@ public class ByteStreamStateHandle implements StreamStateHandle {
public int read() throws IOException {
return index < data.length ? data[index++] & 0xFF : -1;
}
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ // note that any bounds checking on "byte[] b" happend anyways by the
+ // System.arraycopy() call below, so we don't add extra checks here
+
+ final int bytesLeft = data.length - index;
+ if (bytesLeft > 0) {
+ final int bytesToCopy = Math.min(len, bytesLeft);
+ System.arraycopy(data, index, b, off, bytesToCopy);
+ index += bytesToCopy;
+ return bytesToCopy;
+ }
+ else {
+ return -1;
+ }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1d38e0b4/flink-runtime/src/test/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandleTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandleTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandleTest.java
new file mode 100644
index 0000000..0175dc3
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/memory/ByteStreamStateHandleTest.java
@@ -0,0 +1,177 @@
+/*
+ * 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.state.memory;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link ByteStreamStateHandle}.
+ */
+public class ByteStreamStateHandleTest {
+
+ @Test
+ public void testStreamSeekAndPos() throws IOException {
+ final byte[] data = {34, 25, 22, 66, 88, 54};
+ final ByteStreamStateHandle handle = new ByteStreamStateHandle("name", data);
+
+ // read backwards, one byte at a time
+ for (int i = data.length; i >= 0; i--) {
+ FSDataInputStream in = handle.openInputStream();
+ in.seek(i);
+
+ assertEquals(i, (int) in.getPos());
+
+ if (i < data.length) {
+ assertEquals((int) data[i], in.read());
+ assertEquals(i + 1, (int) in.getPos());
+ } else {
+ assertEquals(-1, in.read());
+ assertEquals(i, (int) in.getPos());
+ }
+ }
+
+ // reading past the end makes no difference
+ FSDataInputStream in = handle.openInputStream();
+ in.seek(data.length);
+
+ // read multiple times, should not affect anything
+ assertEquals(-1, in.read());
+ assertEquals(-1, in.read());
+ assertEquals(-1, in.read());
+
+ assertEquals(data.length, (int) in.getPos());
+ }
+
+ @Test
+ public void testStreamSeekOutOfBounds() throws IOException {
+ final int len = 10;
+ final ByteStreamStateHandle handle = new ByteStreamStateHandle("name", new byte[len]);
+
+ // check negative offset
+ FSDataInputStream in = handle.openInputStream();
+ try {
+ in.seek(-2);
+ fail("should fail with an exception");
+ } catch (IOException e) {
+ // expected
+ }
+
+ // check integer overflow
+ in = handle.openInputStream();
+ try {
+ in.seek(len + 1);
+ fail("should fail with an exception");
+ } catch (IOException e) {
+ // expected
+ }
+
+ // check integer overflow
+ in = handle.openInputStream();
+ try {
+ in.seek(((long) Integer.MAX_VALUE) + 100L);
+ fail("should fail with an exception");
+ } catch (IOException e) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testBulkRead() throws IOException {
+ final byte[] data = {34, 25, 22, 66};
+ final ByteStreamStateHandle handle = new ByteStreamStateHandle("name", data);
+ final int targetLen = 8;
+
+ for (int start = 0; start < data.length; start++) {
+ for (int num = 0; num < targetLen; num++) {
+ FSDataInputStream in = handle.openInputStream();
+ in.seek(start);
+
+ final byte[] target = new byte[targetLen];
+ final int read = in.read(target, targetLen - num, num);
+
+ assertEquals(Math.min(num, data.length - start), read);
+ for (int i = 0; i < read; i++) {
+ assertEquals(data[start + i], target[targetLen - num + i]);
+ }
+
+ int newPos = start + read;
+ assertEquals(newPos, (int) in.getPos());
+ assertEquals(newPos < data.length ? data[newPos] : -1, in.read());
+ }
+ }
+ }
+
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @Test
+ public void testBulkReadINdexOutOfBounds() throws IOException {
+ final ByteStreamStateHandle handle = new ByteStreamStateHandle("name", new byte[10]);
+
+ // check negative offset
+ FSDataInputStream in = handle.openInputStream();
+ try {
+ in.read(new byte[10], -1, 5);
+ fail("should fail with an exception");
+ } catch (IndexOutOfBoundsException e) {
+ // expected
+ }
+
+ // check offset overflow
+ in = handle.openInputStream();
+ try {
+ in.read(new byte[10], 10, 5);
+ fail("should fail with an exception");
+ } catch (IndexOutOfBoundsException e) {
+ // expected
+ }
+
+ // check negative length
+ in = handle.openInputStream();
+ try {
+ in.read(new byte[10], 0, -2);
+ fail("should fail with an exception");
+ } catch (IndexOutOfBoundsException e) {
+ // expected
+ }
+
+ // check length too large
+ in = handle.openInputStream();
+ try {
+ in.read(new byte[10], 5, 6);
+ fail("should fail with an exception");
+ } catch (IndexOutOfBoundsException e) {
+ // expected
+ }
+
+ // check length integer overflow
+ in = handle.openInputStream();
+ try {
+ in.read(new byte[10], 5, Integer.MAX_VALUE);
+ fail("should fail with an exception");
+ } catch (IndexOutOfBoundsException e) {
+ // expected
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/1d38e0b4/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateInitializationContextImplTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateInitializationContextImplTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateInitializationContextImplTest.java
index 1ba2e77..700f7e1 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateInitializationContextImplTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateInitializationContextImplTest.java
@@ -38,7 +38,6 @@ import org.apache.flink.runtime.state.StateInitializationContextImpl;
import org.apache.flink.runtime.state.StatePartitionStreamProvider;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
import org.apache.flink.runtime.util.LongArrayList;
-import org.apache.flink.util.Preconditions;
import org.junit.Assert;
import org.junit.Before;
@@ -251,19 +250,20 @@ public class StateInitializationContextImplTest {
@Override
public FSDataInputStream openInputStream() throws IOException {
+ final FSDataInputStream original = super.openInputStream();
+
return new FSDataInputStream() {
- private int index = 0;
+
private boolean closed = false;
@Override
public void seek(long desired) throws IOException {
- Preconditions.checkArgument(desired >= 0 && desired < Integer.MAX_VALUE);
- index = (int) desired;
+ original.seek(desired);
}
@Override
public long getPos() throws IOException {
- return index;
+ return original.getPos();
}
@Override
@@ -271,12 +271,12 @@ public class StateInitializationContextImplTest {
if (closed) {
throw new IOException("Stream closed");
}
- return index < data.length ? data[index++] & 0xFF : -1;
+ return original.read();
}
@Override
public void close() throws IOException {
- super.close();
+ original.close();
this.closed = true;
}
};