You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/07/29 12:01:57 UTC
[flink] 02/05: [FLINK-23452][runtime] Added extra classes for the
ability to calculate throughput
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fe26c9d0644ce050b50c1cf14a5c917f07d006ca
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Tue Jul 27 18:30:22 2021 +0200
[FLINK-23452][runtime] Added extra classes for the ability to calculate throughput
---
.../runtime/throughput/ThroughputCalculator.java | 74 ++++++++++++
.../flink/runtime/throughput/ThroughputEMA.java | 63 ++++++++++
.../throughput/ThroughputCalculatorTest.java | 128 +++++++++++++++++++++
.../runtime/throughput/ThroughputEMATest.java | 93 +++++++++++++++
4 files changed, 358 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/ThroughputCalculator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/ThroughputCalculator.java
new file mode 100644
index 0000000..83aa408
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/ThroughputCalculator.java
@@ -0,0 +1,74 @@
+/*
+ * 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.throughput;
+
+import org.apache.flink.util.clock.Clock;
+
+/** Class for measuring the throughput based on incoming data size and measurement period. */
+public class ThroughputCalculator {
+ private static final long NOT_TRACKED = -1;
+ private final Clock clock;
+ private final ThroughputEMA throughputEMA;
+
+ private long currentAccumulatedDataSize;
+ private long currentMeasurementTime;
+ private long measurementStartTime = NOT_TRACKED;
+ private long lastThroughput;
+
+ public ThroughputCalculator(Clock clock, int numberOfSamples) {
+ this.clock = clock;
+ this.throughputEMA = new ThroughputEMA(numberOfSamples);
+ }
+
+ public void incomingDataSize(long receivedDataSize) {
+ resumeMeasurement();
+ currentAccumulatedDataSize += receivedDataSize;
+ }
+
+ /** Mark when the time should not be taken into account. */
+ public void pauseMeasurement() {
+ if (measurementStartTime != NOT_TRACKED) {
+ currentMeasurementTime += clock.relativeTimeMillis() - measurementStartTime;
+ }
+ measurementStartTime = NOT_TRACKED;
+ }
+
+ /** Mark when the time should be included to the throughput calculation. */
+ public void resumeMeasurement() {
+ if (measurementStartTime == NOT_TRACKED) {
+ measurementStartTime = clock.relativeTimeMillis();
+ }
+ }
+
+ /** @return Calculated throughput based on the collected data for the last period. */
+ public long calculateThroughput() {
+ if (measurementStartTime != NOT_TRACKED) {
+ currentMeasurementTime += clock.relativeTimeMillis() - measurementStartTime;
+ }
+
+ lastThroughput =
+ throughputEMA.calculateThroughput(
+ currentAccumulatedDataSize, currentMeasurementTime);
+
+ measurementStartTime = clock.relativeTimeMillis();
+ currentAccumulatedDataSize = currentMeasurementTime = 0;
+
+ return lastThroughput;
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/ThroughputEMA.java b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/ThroughputEMA.java
new file mode 100644
index 0000000..30d066f
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/ThroughputEMA.java
@@ -0,0 +1,63 @@
+/*
+ * 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.throughput;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Implementation of 'Exponential moving average' algorithm. */
+public class ThroughputEMA {
+ private static final long MILLIS_IN_SECOND = 1000;
+ private long currentThroughput;
+
+ /** EMA algorithm specific constant which responsible for speed of reaction. */
+ private final double alpha;
+
+ public ThroughputEMA(long numberOfSamples) {
+ alpha = 2.0 / (numberOfSamples + 1);
+ }
+
+ /**
+ * Calculating the throughput over the size of data which were received in given amount of time.
+ *
+ * @param dataSize Size of the data which were received in given time.
+ * @param time Time during which the data was received.
+ * @return Throughput calculated according to implemented algorithm.
+ */
+ public long calculateThroughput(long dataSize, long time) {
+ checkArgument(dataSize >= 0, "Size of data should be non negative");
+ checkArgument(time >= 0, "Time should be non negative");
+
+ if (time == 0) {
+ return currentThroughput;
+ }
+
+ if (currentThroughput != 0) {
+ currentThroughput += alpha * (instantThroughput(dataSize, time) - currentThroughput);
+ } else {
+ // Warm-up case.
+ currentThroughput = instantThroughput(dataSize, time);
+ }
+
+ return currentThroughput;
+ }
+
+ static long instantThroughput(long dataSize, long time) {
+ return (long) ((double) dataSize / time * MILLIS_IN_SECOND);
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/ThroughputCalculatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/ThroughputCalculatorTest.java
new file mode 100644
index 0000000..dbe537d
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/ThroughputCalculatorTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.throughput;
+
+import org.apache.flink.util.clock.ManualClock;
+
+import junit.framework.TestCase;
+import org.junit.Test;
+
+import java.time.Duration;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+/** Test for {@link ThroughputCalculator}. */
+public class ThroughputCalculatorTest extends TestCase {
+
+ @Test
+ public void testCorrectThroughputCalculation() {
+ ManualClock clock = new ManualClock();
+ ThroughputCalculator throughputCalculator = new ThroughputCalculator(clock, 10);
+ // Start throughput time.
+ throughputCalculator.calculateThroughput();
+
+ throughputCalculator.incomingDataSize(6666);
+ clock.advanceTime(Duration.ofMillis(1));
+ throughputCalculator.incomingDataSize(3333);
+ clock.advanceTime(Duration.ofMillis(33));
+ throughputCalculator.incomingDataSize(1);
+ clock.advanceTime(Duration.ofMillis(66));
+
+ assertThat(throughputCalculator.calculateThroughput(), is(10_000L * 1_000 / 100));
+ }
+
+ @Test
+ public void testResetValueAfterCalculation() {
+ ManualClock clock = new ManualClock();
+ ThroughputCalculator throughputCalculator = new ThroughputCalculator(clock, 10);
+ // Start throughput time.
+ throughputCalculator.calculateThroughput();
+
+ throughputCalculator.incomingDataSize(666);
+ clock.advanceTime(Duration.ofMillis(100));
+
+ assertThat(throughputCalculator.calculateThroughput(), is(6660L));
+ // It should be the same as previous time.
+ assertThat(throughputCalculator.calculateThroughput(), is(6660L));
+
+ clock.advanceTime(Duration.ofMillis(1));
+ assertThat(throughputCalculator.calculateThroughput(), is(5449L));
+ }
+
+ @Test
+ public void testIgnoringIdleTime() {
+ ManualClock clock = new ManualClock();
+ ThroughputCalculator throughputCalculator = new ThroughputCalculator(clock, 10);
+ // Start throughput time.
+ throughputCalculator.calculateThroughput();
+
+ throughputCalculator.incomingDataSize(7);
+ clock.advanceTime(Duration.ofMillis(1));
+ throughputCalculator.pauseMeasurement();
+ // This will be ignored because it is in idle now.
+ clock.advanceTime(Duration.ofMillis(9));
+ // This should resume the measurement time.
+ throughputCalculator.incomingDataSize(3);
+ clock.advanceTime(Duration.ofMillis(1));
+
+ assertThat(throughputCalculator.calculateThroughput(), is(5L * 1_000));
+ }
+
+ @Test
+ public void testCalculationDuringIdleTime() {
+ ManualClock clock = new ManualClock();
+ ThroughputCalculator throughputCalculator = new ThroughputCalculator(clock, 10);
+ // Start throughput time.
+ throughputCalculator.calculateThroughput();
+
+ throughputCalculator.incomingDataSize(10);
+ clock.advanceTime(Duration.ofMillis(1));
+ throughputCalculator.pauseMeasurement();
+ // This will be ignored because it is in idle now.
+ clock.advanceTime(Duration.ofMillis(9));
+
+ assertThat(throughputCalculator.calculateThroughput(), is(10L * 1_000));
+ }
+
+ @Test
+ public void testMultiplyIdleEnd() {
+ ManualClock clock = new ManualClock();
+ ThroughputCalculator throughputCalculator = new ThroughputCalculator(clock, 10);
+
+ // Start throughput time.
+ throughputCalculator.calculateThroughput();
+
+ throughputCalculator.incomingDataSize(10);
+ // It won't be ignored.
+ clock.advanceTime(Duration.ofMillis(3));
+ throughputCalculator.resumeMeasurement();
+ // It won't be ignored.
+ clock.advanceTime(Duration.ofMillis(3));
+ throughputCalculator.resumeMeasurement();
+ // It won't be ignored.
+ clock.advanceTime(Duration.ofMillis(3));
+ throughputCalculator.resumeMeasurement();
+
+ clock.advanceTime(Duration.ofMillis(1));
+
+ // resumeMeasurement should not reset the time because pauseMeasurement was not called.
+ assertThat(throughputCalculator.calculateThroughput(), is(1_000L));
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/ThroughputEMATest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/ThroughputEMATest.java
new file mode 100644
index 0000000..94df20a
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/ThroughputEMATest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.throughput;
+
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+/** Test for {@link ThroughputEMA}. */
+public class ThroughputEMATest extends TestLogger {
+ @Test
+ public void testWarmUpThroughputCalculation() {
+ ThroughputEMA calculator = new ThroughputEMA(5);
+
+ // Calculation throughput bytes/second.
+ assertThat(calculator.calculateThroughput(100, 33), is(3030L));
+ assertThat(calculator.calculateThroughput(200, 33), is(4040L));
+ assertThat(calculator.calculateThroughput(100, 9), is(6397L));
+ }
+
+ @Test
+ public void testWarmUpIsFinished() {
+ ThroughputEMA calculator = new ThroughputEMA(2);
+
+ assertThat(calculator.calculateThroughput(100, 33), is(3030L));
+ assertThat(calculator.calculateThroughput(200, 33), is(5050L));
+ }
+
+ @Test
+ public void testCalculationThroughput() {
+ ThroughputEMA calculator = new ThroughputEMA(3);
+
+ // warm-up
+ assertThat(calculator.calculateThroughput(100, 33), is(3030L));
+ assertThat(calculator.calculateThroughput(200, 33), is(4545L));
+ assertThat(calculator.calculateThroughput(100, 10), is(7272L));
+
+ // Calculation of the average
+ assertThat(calculator.calculateThroughput(100, 10), is(8636L));
+
+ // The result value seeks to the upper limit but it will take a while until it reaches it.
+ assertThat(calculator.calculateThroughput(100, 10), is(9318L));
+ assertThat(calculator.calculateThroughput(100, 10), is(9659L));
+ assertThat(calculator.calculateThroughput(100, 10), is(9829L));
+ assertThat(calculator.calculateThroughput(100, 10), is(9914L));
+ assertThat(calculator.calculateThroughput(100, 10), is(9957L));
+
+ // Downgrade
+ // The fast drop for the first value because the last value has the highest weight.
+ assertThat(calculator.calculateThroughput(10, 10), is(5478L));
+ assertThat(calculator.calculateThroughput(10, 10), is(3239L));
+ assertThat(calculator.calculateThroughput(10, 10), is(2119L));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testNegativeDataSize() {
+ ThroughputEMA calculator = new ThroughputEMA(2);
+ calculator.calculateThroughput(-1, 1);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testNegativeTime() {
+ ThroughputEMA calculator = new ThroughputEMA(2);
+ calculator.calculateThroughput(1, -1);
+ }
+
+ @Test
+ public void testPreviousValueWhenTimeIsZero() {
+ ThroughputEMA calculator = new ThroughputEMA(2);
+
+ assertThat(calculator.calculateThroughput(100, 33), is(3030L));
+ assertThat(calculator.calculateThroughput(200, 0), is(3030L));
+ }
+}