This is an automated email from the ASF dual-hosted git repository. dwysakowicz pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink-benchmarks.git
The following commit(s) were added to refs/heads/master by this push: new 325b1b0 [FLINK-24231] Benchmark for buffer debloating multiple gates 325b1b0 is described below commit 325b1b0d94d00fb4f37152d89435e3485edf3004 Author: Dawid Wysakowicz <dwysakow...@apache.org> AuthorDate: Wed Oct 6 19:36:39 2021 +0200 [FLINK-24231] Benchmark for buffer debloating multiple gates --- .../apache/flink/benchmark/BackpressureUtils.java | 87 +++++++++ .../benchmark/CheckpointEnvironmentContext.java | 175 +++++++++++++++++ .../benchmark/CheckpointingTimeBenchmark.java | 210 ++++----------------- .../MultiInputCheckpointingTimeBenchmark.java | 138 ++++++++++++++ 4 files changed, 432 insertions(+), 178 deletions(-) diff --git a/src/main/java/org/apache/flink/benchmark/BackpressureUtils.java b/src/main/java/org/apache/flink/benchmark/BackpressureUtils.java new file mode 100644 index 0000000..cc39f01 --- /dev/null +++ b/src/main/java/org/apache/flink/benchmark/BackpressureUtils.java @@ -0,0 +1,87 @@ +/* + * 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.benchmark; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.time.Deadline; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.UnmodifiableConfiguration; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.rest.RestClient; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobVertexBackPressureHeaders; +import org.apache.flink.runtime.rest.messages.JobVertexBackPressureInfo; +import org.apache.flink.runtime.rest.messages.JobVertexMessageParameters; +import org.apache.flink.util.FlinkRuntimeException; +import org.apache.flink.util.concurrent.ExecutorThreadFactory; + +import java.net.URI; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.Executors; + +/** + * Utility class for querying a backpressure status. + */ +public class BackpressureUtils { + static void waitForBackpressure(JobID jobID, List<JobVertexID> sourceId, URI restAddress) + throws Exception { + final RestClient restClient = + new RestClient( + new UnmodifiableConfiguration(new Configuration()), + Executors.newSingleThreadScheduledExecutor( + new ExecutorThreadFactory("Flink-RestClient-IO"))); + Deadline deadline = Deadline.fromNow(Duration.ofSeconds(30)); + boolean allBackpressured; + do { + allBackpressured = + sourceId.stream() + .map(id -> queryBackpressure(jobID, id, restClient, restAddress)) + .allMatch( + level -> + level + == JobVertexBackPressureInfo + .VertexBackPressureLevel.HIGH); + } while (!allBackpressured && deadline.hasTimeLeft()); + if (!allBackpressured) { + throw new FlinkRuntimeException( + "Could not trigger backpressure for the job in given time."); + } + } + + private static JobVertexBackPressureInfo.VertexBackPressureLevel queryBackpressure( + JobID jobID, JobVertexID vertexID, RestClient restClient, URI restAddress) { + try { + final JobVertexMessageParameters metricsParameters = new JobVertexMessageParameters(); + metricsParameters.jobPathParameter.resolve(jobID); + metricsParameters.jobVertexIdPathParameter.resolve(vertexID); + return restClient + .sendRequest( + restAddress.getHost(), + restAddress.getPort(), + JobVertexBackPressureHeaders.getInstance(), + metricsParameters, + EmptyRequestBody.getInstance()) + .get() + .getBackpressureLevel(); + } catch (Exception e) { + throw new FlinkRuntimeException(e); + } + } +} diff --git a/src/main/java/org/apache/flink/benchmark/CheckpointEnvironmentContext.java b/src/main/java/org/apache/flink/benchmark/CheckpointEnvironmentContext.java new file mode 100644 index 0000000..085b53e --- /dev/null +++ b/src/main/java/org/apache/flink/benchmark/CheckpointEnvironmentContext.java @@ -0,0 +1,175 @@ +/* + * 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.benchmark; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.MemorySize; +import org.apache.flink.configuration.TaskManagerOptions; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.testutils.CommonTestUtils; +import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.apache.flink.streaming.api.graph.StreamGraph; + +import java.time.Duration; +import java.time.temporal.ChronoUnit; +import java.util.List; +import java.util.function.Function; + +/** + * Common context to be used for benchmarking checkpointing time. + * + * @see CheckpointingTimeBenchmark + * @see MultiInputCheckpointingTimeBenchmark + */ +public abstract class CheckpointEnvironmentContext extends FlinkEnvironmentContext { + + public static final int JOB_PARALLELISM = 4; + public static final MemorySize START_MEMORY_SEGMENT_SIZE = MemorySize.parse("8 kb"); + public static final MemorySize MIN_MEMORY_SEGMENT_SIZE = MemorySize.parse("256 b"); + public static final Duration DEBLOATING_TARGET = Duration.of(300, ChronoUnit.MILLIS); + public static final int DEBLOATING_STABILIZATION_PERIOD = 2_000; + + public JobID jobID; + + /** + * Checkpointing configuration to be used in {@link CheckpointingTimeBenchmark} & {@link + * MultiInputCheckpointingTimeBenchmark}. + */ + public enum CheckpointMode { + UNALIGNED( + config -> { + config.set(ExecutionCheckpointingOptions.ENABLE_UNALIGNED, true); + config.set( + TaskManagerOptions.MEMORY_SEGMENT_SIZE, + CheckpointEnvironmentContext.START_MEMORY_SEGMENT_SIZE); + config.set( + ExecutionCheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT, + Duration.ofMillis(0)); + config.set(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED, false); + return config; + }), + UNALIGNED_1( + config -> { + config.set(ExecutionCheckpointingOptions.ENABLE_UNALIGNED, true); + config.set( + TaskManagerOptions.MEMORY_SEGMENT_SIZE, + CheckpointEnvironmentContext.START_MEMORY_SEGMENT_SIZE); + config.set( + ExecutionCheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT, + Duration.ofMillis(1)); + config.set(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED, false); + return config; + }), + ALIGNED( + config -> { + config.set(ExecutionCheckpointingOptions.ENABLE_UNALIGNED, false); + config.set( + TaskManagerOptions.MEMORY_SEGMENT_SIZE, + CheckpointEnvironmentContext.START_MEMORY_SEGMENT_SIZE); + config.set( + TaskManagerOptions.MIN_MEMORY_SEGMENT_SIZE, + CheckpointEnvironmentContext.MIN_MEMORY_SEGMENT_SIZE); + config.set(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED, true); + config.set( + TaskManagerOptions.BUFFER_DEBLOAT_TARGET, + CheckpointEnvironmentContext.DEBLOATING_TARGET); + config.set( + TaskManagerOptions.BUFFER_DEBLOAT_PERIOD, + Duration.of(200, ChronoUnit.MILLIS)); + config.set(TaskManagerOptions.BUFFER_DEBLOAT_SAMPLES, 5); + return config; + }); + private final Function<Configuration, Configuration> configFunc; + + CheckpointMode(Function<Configuration, Configuration> configFunc) { + this.configFunc = configFunc; + } + + public Configuration configure(Configuration config) { + return configFunc.apply(config); + } + } + + @Override + public void setUp() throws Exception { + super.setUp(); + env.setParallelism(CheckpointEnvironmentContext.JOB_PARALLELISM); + env.enableCheckpointing(Long.MAX_VALUE); + + final StreamGraphWithSources streamGraphWithSources = getStreamGraph(); + + final JobClient jobClient = env.executeAsync(streamGraphWithSources.getStreamGraph()); + jobID = jobClient.getJobID(); + CommonTestUtils.waitForAllTaskRunning(miniCluster, jobID, false); + BackpressureUtils.waitForBackpressure( + jobID, streamGraphWithSources.getSources(), miniCluster.getRestAddress().get()); + if (getSleepPostSetUp() > 0) { + Thread.sleep(getSleepPostSetUp()); + } + } + + protected abstract CheckpointMode getMode(); + + protected abstract StreamGraphWithSources getStreamGraph(); + + protected int getSleepPostSetUp() { + return getMode() == CheckpointMode.ALIGNED + ? CheckpointEnvironmentContext.DEBLOATING_STABILIZATION_PERIOD + : 0; + } + + @Override + protected Configuration createConfiguration() { + return getMode().configure(super.createConfiguration()); + } + + /** A simple wrapper to pass a {@link StreamGraph} along with ids of sources it contains. */ + public static class StreamGraphWithSources { + private final StreamGraph streamGraph; + private final List<JobVertexID> sources; + + public StreamGraphWithSources(StreamGraph streamGraph, List<JobVertexID> sources) { + this.streamGraph = streamGraph; + this.sources = sources; + } + + public StreamGraph getStreamGraph() { + return streamGraph; + } + + public List<JobVertexID> getSources() { + return sources; + } + } + + /** + * The custom sink for processing records slowly to cause accumulate in-flight buffers even back + * pressure. + */ + public static class SlowDiscardSink<T> implements SinkFunction<T> { + @Override + public void invoke(T value, Context context) { + final long startTime = System.nanoTime(); + while (System.nanoTime() - startTime < 200_000) {} + } + } +} diff --git a/src/main/java/org/apache/flink/benchmark/CheckpointingTimeBenchmark.java b/src/main/java/org/apache/flink/benchmark/CheckpointingTimeBenchmark.java index 7b93900..a56b4fa 100644 --- a/src/main/java/org/apache/flink/benchmark/CheckpointingTimeBenchmark.java +++ b/src/main/java/org/apache/flink/benchmark/CheckpointingTimeBenchmark.java @@ -20,46 +20,26 @@ package org.apache.flink.benchmark; import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.functions.MapFunction; -import org.apache.flink.api.common.time.Deadline; import org.apache.flink.benchmark.operators.RecordSource; -import org.apache.flink.benchmark.operators.RecordSource.Record; -import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.MemorySize; -import org.apache.flink.configuration.TaskManagerOptions; -import org.apache.flink.configuration.UnmodifiableConfiguration; -import org.apache.flink.core.execution.JobClient; import org.apache.flink.runtime.jobgraph.JobVertexID; -import org.apache.flink.runtime.rest.RestClient; -import org.apache.flink.runtime.rest.messages.EmptyRequestBody; -import org.apache.flink.runtime.rest.messages.JobVertexBackPressureHeaders; -import org.apache.flink.runtime.rest.messages.JobVertexBackPressureInfo; -import org.apache.flink.runtime.rest.messages.JobVertexMessageParameters; -import org.apache.flink.runtime.testutils.CommonTestUtils; import org.apache.flink.streaming.api.datastream.DataStreamSource; -import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions; import org.apache.flink.streaming.api.functions.sink.SinkFunction; -import org.apache.flink.util.FlinkRuntimeException; -import org.apache.flink.util.concurrent.ExecutorThreadFactory; +import org.apache.flink.streaming.api.graph.StreamGraph; import org.openjdk.jmh.annotations.Benchmark; -import org.openjdk.jmh.annotations.Measurement; import org.openjdk.jmh.annotations.OutputTimeUnit; import org.openjdk.jmh.annotations.Param; import org.openjdk.jmh.annotations.Scope; import org.openjdk.jmh.annotations.State; -import org.openjdk.jmh.annotations.Warmup; import org.openjdk.jmh.runner.Runner; import org.openjdk.jmh.runner.RunnerException; import org.openjdk.jmh.runner.options.Options; import org.openjdk.jmh.runner.options.OptionsBuilder; import org.openjdk.jmh.runner.options.VerboseMode; -import java.net.URI; -import java.time.Duration; -import java.time.temporal.ChronoUnit; +import java.util.Collections; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.Executors; -import java.util.function.Function; import static java.util.concurrent.TimeUnit.SECONDS; import static org.apache.flink.api.common.eventtime.WatermarkStrategy.noWatermarks; @@ -73,25 +53,20 @@ import static org.apache.flink.api.common.eventtime.WatermarkStrategy.noWatermar * <ul> * <li>The minimal memory segment size is decreased (256b) so that the scaling possibility is * higher. Memory segments start with 8kb - * <li>A memory segment of the minimal size fits ~9 records (of size 29b), each record takes ~200ns - * to be processed by the sink + * <li>A memory segment of the minimal size fits ~9 records (of size 29b), each record takes + * ~200ns to be processed by the sink * <li>We have 2 (exclusive buffers) * 4 (parallelism) + 8 floating = 64 buffers per gate, with * 300 ms debloating target and ~200ns/record processing speed, we can buffer 1500/64 = ~24 - * records in a buffer after debloating which means the size of a buffer (24 * 29 = 696) is slightly above the - * minimal memory segment size. + * records in a buffer after debloating which means the size of a buffer (24 * 29 = 696) is + * slightly above the minimal memory segment size. * <li>The buffer debloating target of 300ms means a checkpoint should take ~2(number of * exchanges)*300ms=~600ms * </ul> */ @OutputTimeUnit(SECONDS) public class CheckpointingTimeBenchmark extends BenchmarkBase { - public static final int JOB_PARALLELISM = 4; - public static final MemorySize START_MEMORY_SEGMENT_SIZE = MemorySize.parse("8 kb"); - public static final MemorySize MIN_MEMORY_SEGMENT_SIZE = MemorySize.parse("256 b"); - public static final Duration DEBLOATING_TARGET = Duration.of(300, ChronoUnit.MILLIS); public static final MemorySize DEBLOATING_RECORD_SIZE = MemorySize.parse("1b"); public static final MemorySize UNALIGNED_RECORD_SIZE = MemorySize.parse("1kb"); - public static final int DEBLOATING_STABILIZATION_PERIOD = 2_000; public static void main(String[] args) throws RunnerException { Options options = @@ -104,182 +79,61 @@ public class CheckpointingTimeBenchmark extends BenchmarkBase { } @Benchmark - public void checkpointSingleInput(CheckpointEnvironmentContext context) throws Exception { + public void checkpointSingleInput(SingleInputCheckpointEnvironmentContext context) + throws Exception { final CompletableFuture<String> checkpoint = context.miniCluster.triggerCheckpoint(context.jobID); checkpoint.get(); } - public enum CheckpointMode { - UNALIGNED( - config -> { - config.set(ExecutionCheckpointingOptions.ENABLE_UNALIGNED, true); - config.set(TaskManagerOptions.MEMORY_SEGMENT_SIZE, START_MEMORY_SEGMENT_SIZE); - config.set( - ExecutionCheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT, - Duration.ofMillis(0)); - config.set(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED, false); - return config; - }, - 0, - UNALIGNED_RECORD_SIZE), - UNALIGNED_1( - config -> { - config.set(ExecutionCheckpointingOptions.ENABLE_UNALIGNED, true); - config.set(TaskManagerOptions.MEMORY_SEGMENT_SIZE, START_MEMORY_SEGMENT_SIZE); - config.set( - ExecutionCheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT, - Duration.ofMillis(1)); - config.set(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED, false); - return config; - }, - 0, - UNALIGNED_RECORD_SIZE), - ALIGNED( - config -> { - config.set(ExecutionCheckpointingOptions.ENABLE_UNALIGNED, false); - config.set(TaskManagerOptions.MEMORY_SEGMENT_SIZE, START_MEMORY_SEGMENT_SIZE); - config.set(TaskManagerOptions.MIN_MEMORY_SEGMENT_SIZE, MIN_MEMORY_SEGMENT_SIZE); - config.set(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED, true); - config.set(TaskManagerOptions.BUFFER_DEBLOAT_TARGET, DEBLOATING_TARGET); - config.set( - TaskManagerOptions.BUFFER_DEBLOAT_PERIOD, - Duration.of(200, ChronoUnit.MILLIS)); - config.set(TaskManagerOptions.BUFFER_DEBLOAT_SAMPLES, 5); - return config; - }, - DEBLOATING_STABILIZATION_PERIOD, - DEBLOATING_RECORD_SIZE); - - private final Function<Configuration, Configuration> configFunc; - private final int sleepPostSetUp; - private final MemorySize recordSize; - - CheckpointMode( - Function<Configuration, Configuration> configFunc, - int sleepPostSetUp, - MemorySize recordSize) { - this.configFunc = configFunc; - this.sleepPostSetUp = sleepPostSetUp; - this.recordSize = recordSize; - } - - public Configuration configure(Configuration config) { - return configFunc.apply(config); - } - - public MemorySize getRecordSize() { - return recordSize; - } - - public int getSleepPostSetUp() { - return sleepPostSetUp; - } - } - @State(Scope.Thread) - public static class CheckpointEnvironmentContext extends FlinkEnvironmentContext { - public JobID jobID; + public static class SingleInputCheckpointEnvironmentContext + extends CheckpointEnvironmentContext { + public static final int NUM_OF_VERTICES = 3; @Param({"ALIGNED", "UNALIGNED", "UNALIGNED_1"}) public CheckpointMode mode; @Override - public void setUp() throws Exception { - super.setUp(); - env.setParallelism(JOB_PARALLELISM); - env.enableCheckpointing(Long.MAX_VALUE); + protected CheckpointMode getMode() { + return mode; + } - DataStreamSource<Record> source = + @Override + protected StreamGraphWithSources getStreamGraph() { + DataStreamSource<RecordSource.Record> source = env.fromSource( - new RecordSource( - Integer.MAX_VALUE, (int) mode.getRecordSize().getBytes()), + new RecordSource(Integer.MAX_VALUE, (int) getRecordSize().getBytes()), noWatermarks(), RecordSource.class.getName()); source.slotSharingGroup("source") .rebalance() - .map((MapFunction<Record, Record>) value -> value) + .map((MapFunction<RecordSource.Record, RecordSource.Record>) value -> value) .slotSharingGroup("map") .rebalance() .addSink(new SlowDiscardSink<>()) .slotSharingGroup("sink"); - final JobVertexID sourceId = extractSourceId(); - final JobClient jobClient = env.executeAsync(); - jobID = jobClient.getJobID(); - CommonTestUtils.waitForAllTaskRunning(miniCluster, jobID, false); - waitForBackpressure(jobID, sourceId); - if (mode.getSleepPostSetUp() > 0) { - Thread.sleep(mode.getSleepPostSetUp()); - } + final StreamGraph streamGraph = env.getStreamGraph(false); + final JobVertexID sourceId = + streamGraph + .getJobGraph() + .getVerticesSortedTopologicallyFromSources() + .get(0) + .getID(); + return new StreamGraphWithSources(streamGraph, Collections.singletonList(sourceId)); } - private JobVertexID extractSourceId() { - return env.getStreamGraph(false) - .getJobGraph() - .getVerticesSortedTopologicallyFromSources() - .get(0) - .getID(); - } - - private void waitForBackpressure(JobID jobID, JobVertexID sourceId) throws Exception { - final RestClient restClient = - new RestClient( - new UnmodifiableConfiguration(new Configuration()), - Executors.newSingleThreadScheduledExecutor( - new ExecutorThreadFactory("Flink-RestClient-IO"))); - final URI restAddress = miniCluster.getRestAddress().get(); - final JobVertexMessageParameters metricsParameters = new JobVertexMessageParameters(); - metricsParameters.jobPathParameter.resolve(jobID); - metricsParameters.jobVertexIdPathParameter.resolve(sourceId); - JobVertexBackPressureInfo responseBody; - Deadline deadline = Deadline.fromNow(Duration.ofSeconds(30)); - do { - responseBody = - restClient - .sendRequest( - restAddress.getHost(), - restAddress.getPort(), - JobVertexBackPressureHeaders.getInstance(), - metricsParameters, - EmptyRequestBody.getInstance()) - .get(); - } while (responseBody.getBackpressureLevel() - != JobVertexBackPressureInfo.VertexBackPressureLevel.HIGH - && deadline.hasTimeLeft()); - if (responseBody.getBackpressureLevel() - != JobVertexBackPressureInfo.VertexBackPressureLevel.HIGH) { - throw new FlinkRuntimeException( - "Could not trigger backpressure for the job in given time."); - } - } - - @Override - protected Configuration createConfiguration() { - return mode.configure(super.createConfiguration()); + private MemorySize getRecordSize() { + return mode == CheckpointMode.ALIGNED + ? CheckpointingTimeBenchmark.DEBLOATING_RECORD_SIZE + : CheckpointingTimeBenchmark.UNALIGNED_RECORD_SIZE; } @Override protected int getNumberOfTaskManagers() { - return 3 * JOB_PARALLELISM; - } - - @Override - protected int getNumberOfSlotsPerTaskManager() { - return 1; - } - } - - /** - * The custom sink for processing records slowly to cause accumulate in-flight buffers even back - * pressure. - */ - public static class SlowDiscardSink<T> implements SinkFunction<T> { - @Override - public void invoke(T value, Context context) { - final long startTime = System.nanoTime(); - while (System.nanoTime() - startTime < 200_000) {} + return NUM_OF_VERTICES * JOB_PARALLELISM; } } } diff --git a/src/main/java/org/apache/flink/benchmark/MultiInputCheckpointingTimeBenchmark.java b/src/main/java/org/apache/flink/benchmark/MultiInputCheckpointingTimeBenchmark.java new file mode 100644 index 0000000..182478b --- /dev/null +++ b/src/main/java/org/apache/flink/benchmark/MultiInputCheckpointingTimeBenchmark.java @@ -0,0 +1,138 @@ +/* + * 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.benchmark; + +import org.apache.flink.benchmark.operators.RecordSource; +import org.apache.flink.benchmark.operators.RecordSource.Record; +import org.apache.flink.configuration.MemorySize; +import org.apache.flink.runtime.jobgraph.JobGraph; +import org.apache.flink.runtime.jobgraph.JobVertex; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.functions.co.CoMapFunction; +import org.apache.flink.streaming.api.graph.StreamGraph; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.RunnerException; +import org.openjdk.jmh.runner.options.Options; +import org.openjdk.jmh.runner.options.OptionsBuilder; +import org.openjdk.jmh.runner.options.VerboseMode; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.apache.flink.api.common.eventtime.WatermarkStrategy.noWatermarks; + +/** + * The test verifies that the debloating kicks in and properly downsizes buffers in case of multi + * input gates with different throughput. In the end the checkpoint should take ~1(number of + * rebalance) * DEBLOATING_TARGET. + */ +@OutputTimeUnit(SECONDS) +public class MultiInputCheckpointingTimeBenchmark extends BenchmarkBase { + + public static final MemorySize SMALL_RECORD_SIZE = MemorySize.parse("1b"); + public static final MemorySize BIG_RECORD_SIZE = MemorySize.parse("1kb"); + + public static void main(String[] args) throws RunnerException { + Options options = + new OptionsBuilder() + .verbosity(VerboseMode.NORMAL) + .include(MultiInputCheckpointingTimeBenchmark.class.getCanonicalName()) + .build(); + + new Runner(options).run(); + } + + @Benchmark + public void checkpointMultiInput(MultiInputCheckpointEnvironmentContext context) + throws Exception { + final CompletableFuture<String> checkpoint = + context.miniCluster.triggerCheckpoint(context.jobID); + checkpoint.get(); + } + + @State(Scope.Thread) + public static class MultiInputCheckpointEnvironmentContext + extends CheckpointEnvironmentContext { + + private static final int NUM_OF_VERTICES = 3; + + @Override + protected CheckpointMode getMode() { + return CheckpointMode.ALIGNED; + } + + @Override + protected StreamGraphWithSources getStreamGraph() { + DataStream<Record> source1 = + env.fromSource( + new RecordSource( + Integer.MAX_VALUE, (int) SMALL_RECORD_SIZE.getBytes()), + noWatermarks(), + RecordSource.class.getName()) + .slotSharingGroup("source-small-records") + .rebalance(); + + DataStream<Record> source2 = + env.fromSource( + new RecordSource( + Integer.MAX_VALUE, (int) BIG_RECORD_SIZE.getBytes()), + noWatermarks(), + RecordSource.class.getName()) + .slotSharingGroup("source-big-records") + .rebalance(); + + source1.connect(source2) + .map( + new CoMapFunction<Record, Record, Record>() { + @Override + public Record map1(Record record) throws Exception { + return record; + } + + @Override + public Record map2(Record record) throws Exception { + return record; + } + }) + .name("co-map") + .slotSharingGroup("map-and-sink") + .addSink(new SlowDiscardSink<>()) + .slotSharingGroup("map-and-sink"); + + final StreamGraph streamGraph = env.getStreamGraph(false); + final JobGraph jobGraph = streamGraph.getJobGraph(); + final List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources(); + + return new StreamGraphWithSources( + streamGraph, Arrays.asList(vertices.get(0).getID(), vertices.get(1).getID())); + } + + @Override + protected int getNumberOfTaskManagers() { + return NUM_OF_VERTICES * JOB_PARALLELISM; + } + } +}