You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2021/03/09 05:45:10 UTC
[flink-benchmarks] branch master updated: [FLINK-21526] Replace
scheduler benchmarks with wrapper classes
This is an automated email from the ASF dual-hosted git repository.
zhuzh 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 00b1f28 [FLINK-21526] Replace scheduler benchmarks with wrapper classes
00b1f28 is described below
commit 00b1f28562d6a58332c0eda2a2e58e8318aa0478
Author: Thesharing <cy...@outlook.com>
AuthorDate: Wed Mar 3 19:10:23 2021 +0800
[FLINK-21526] Replace scheduler benchmarks with wrapper classes
---
.../scheduler/benchmark/JobConfiguration.java | 78 ---------
.../benchmark/SchedulerBenchmarkUtils.java | 174 ---------------------
...ownstreamTasksInBatchJobBenchmarkExecutor.java} | 31 ++--
.../deploying/DeployingTasksBenchmarkBase.java | 74 ---------
...oyingTasksInStreamingJobBenchmarkExecutor.java} | 20 ++-
.../benchmark/failover/FailoverBenchmarkBase.java | 56 -------
...egionToRestartInBatchJobBenchmarkExecutor.java} | 37 ++---
...nToRestartInStreamingJobBenchmarkExecutor.java} | 30 ++--
.../PartitionReleaseInBatchJobBenchmark.java | 83 ----------
...rtitionReleaseInBatchJobBenchmarkExecutor.java} | 27 ++--
...> InitSchedulingStrategyBenchmarkExecutor.java} | 20 +--
.../scheduling/SchedulingBenchmarkBase.java | 49 ------
...hedulingDownstreamTasksInBatchJobBenchmark.java | 80 ----------
...ownstreamTasksInBatchJobBenchmarkExecutor.java} | 25 +--
...a => BuildExecutionGraphBenchmarkExecutor.java} | 31 ++--
15 files changed, 86 insertions(+), 729 deletions(-)
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/JobConfiguration.java b/src/main/java/org/apache/flink/scheduler/benchmark/JobConfiguration.java
deleted file mode 100644
index 15220ad..0000000
--- a/src/main/java/org/apache/flink/scheduler/benchmark/JobConfiguration.java
+++ /dev/null
@@ -1,78 +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.scheduler.benchmark;
-
-import org.apache.flink.api.common.ExecutionMode;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
-import org.apache.flink.runtime.jobgraph.DistributionPattern;
-import org.apache.flink.runtime.jobgraph.JobType;
-
-/**
- * {@link JobConfiguration} contains the configuration of a STREAMING/BATCH job.
- * It concludes {@link DistributionPattern}, {@link ResultPartitionType}, {@link JobType}, {@link ExecutionMode}.
- */
-public enum JobConfiguration {
-
- STREAMING(DistributionPattern.ALL_TO_ALL,
- ResultPartitionType.PIPELINED,
- JobType.STREAMING,
- ExecutionMode.PIPELINED),
-
- BATCH(DistributionPattern.ALL_TO_ALL,
- ResultPartitionType.BLOCKING,
- JobType.BATCH,
- ExecutionMode.BATCH);
-
- private final static int PARALLELISM = 4000;
- private final DistributionPattern distributionPattern;
- private final ResultPartitionType resultPartitionType;
- private final JobType jobType;
- private final ExecutionMode executionMode;
-
- JobConfiguration(
- DistributionPattern distributionPattern,
- ResultPartitionType resultPartitionType,
- JobType jobType,
- ExecutionMode executionMode) {
- this.distributionPattern = distributionPattern;
- this.resultPartitionType = resultPartitionType;
- this.jobType = jobType;
- this.executionMode = executionMode;
- }
-
- public int getParallelism() {
- return PARALLELISM;
- }
-
- public DistributionPattern getDistributionPattern() {
- return distributionPattern;
- }
-
- public ResultPartitionType getResultPartitionType() {
- return resultPartitionType;
- }
-
- public JobType getJobType() {
- return jobType;
- }
-
- public ExecutionMode getExecutionMode() {
- return executionMode;
- }
-}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/SchedulerBenchmarkUtils.java b/src/main/java/org/apache/flink/scheduler/benchmark/SchedulerBenchmarkUtils.java
deleted file mode 100644
index ba69261..0000000
--- a/src/main/java/org/apache/flink/scheduler/benchmark/SchedulerBenchmarkUtils.java
+++ /dev/null
@@ -1,174 +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.scheduler.benchmark;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.runtime.JobException;
-import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
-import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
-import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.executiongraph.AccessExecutionJobVertex;
-import org.apache.flink.runtime.executiongraph.Execution;
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
-import org.apache.flink.runtime.executiongraph.ExecutionGraph;
-import org.apache.flink.runtime.executiongraph.ExecutionVertex;
-import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
-import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.jobgraph.JobGraphBuilder;
-import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.runtime.jobgraph.JobVertexID;
-import org.apache.flink.runtime.jobmaster.LogicalSlot;
-import org.apache.flink.runtime.jobmaster.TestingLogicalSlotBuilder;
-import org.apache.flink.runtime.scheduler.DefaultScheduler;
-import org.apache.flink.runtime.scheduler.SchedulerTestingUtils;
-import org.apache.flink.runtime.taskmanager.TaskExecutionState;
-import org.apache.flink.runtime.testtasks.NoOpInvokable;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.ExecutionException;
-
-/**
- * Utilities for scheduler benchmarks.
- */
-public class SchedulerBenchmarkUtils {
-
- public static List<JobVertex> createDefaultJobVertices(JobConfiguration jobConfiguration) {
-
- final List<JobVertex> jobVertices = new ArrayList<>();
-
- final JobVertex source = new JobVertex("source");
- source.setInvokableClass(NoOpInvokable.class);
- source.setParallelism(jobConfiguration.getParallelism());
- jobVertices.add(source);
-
- final JobVertex sink = new JobVertex("sink");
- sink.setInvokableClass(NoOpInvokable.class);
- sink.setParallelism(jobConfiguration.getParallelism());
- jobVertices.add(sink);
-
- sink.connectNewDataSetAsInput(
- source,
- jobConfiguration.getDistributionPattern(),
- jobConfiguration.getResultPartitionType());
-
- return jobVertices;
- }
-
- public static JobGraph createJobGraph(JobConfiguration jobConfiguration) throws IOException {
- return createJobGraph(Collections.emptyList(), jobConfiguration);
- }
-
- public static JobGraph createJobGraph(
- List<JobVertex> jobVertices,
- JobConfiguration jobConfiguration) throws IOException {
-
- final JobGraph jobGraph = JobGraphBuilder.newStreamingJobGraphBuilder()
- .addJobVertices(jobVertices)
- .build();
-
- jobGraph.setJobType(jobConfiguration.getJobType());
-
- final ExecutionConfig executionConfig = new ExecutionConfig();
- executionConfig.setExecutionMode(jobConfiguration.getExecutionMode());
- jobGraph.setExecutionConfig(executionConfig);
-
- return jobGraph;
- }
-
- public static ExecutionGraph createAndInitExecutionGraph(
- List<JobVertex> jobVertices,
- JobConfiguration jobConfiguration) throws Exception {
-
- final JobGraph jobGraph = createJobGraph(
- jobVertices,
- jobConfiguration);
-
- final ComponentMainThreadExecutor mainThreadExecutor = ComponentMainThreadExecutorServiceAdapter.forMainThread();
-
- final DefaultScheduler scheduler = SchedulerTestingUtils.createScheduler(jobGraph, mainThreadExecutor);
-
- return scheduler.getExecutionGraph();
- }
-
- public static void deployTasks(
- ExecutionGraph executionGraph,
- JobVertexID jobVertexID,
- TestingLogicalSlotBuilder slotBuilder,
- boolean sendScheduleOrUpdateConsumersMessage) throws JobException, ExecutionException, InterruptedException {
-
- for (ExecutionVertex vertex : executionGraph.getJobVertex(jobVertexID).getTaskVertices()) {
- LogicalSlot slot = slotBuilder.createTestingLogicalSlot();
- Execution execution = vertex.getCurrentExecutionAttempt();
- execution.registerProducedPartitions(
- slot.getTaskManagerLocation(),
- sendScheduleOrUpdateConsumersMessage).get();
- assignResourceAndDeploy(vertex, slot);
- }
- }
-
- public static void deployAllTasks(
- ExecutionGraph executionGraph,
- TestingLogicalSlotBuilder slotBuilder) throws JobException, ExecutionException, InterruptedException {
-
- for (ExecutionVertex vertex : executionGraph.getAllExecutionVertices()) {
- LogicalSlot slot = slotBuilder.createTestingLogicalSlot();
- vertex.getCurrentExecutionAttempt().registerProducedPartitions(slot.getTaskManagerLocation(), true).get();
- assignResourceAndDeploy(vertex, slot);
- }
- }
-
- private static void assignResourceAndDeploy(ExecutionVertex vertex, LogicalSlot slot) throws JobException {
- vertex.tryAssignResource(slot);
- vertex.deploy();
- }
-
- public static void transitionTaskStatus(
- ExecutionGraph executionGraph,
- JobVertexID jobVertexID,
- ExecutionState state) {
-
- for (ExecutionVertex vertex : executionGraph
- .getJobVertex(jobVertexID)
- .getTaskVertices()) {
- executionGraph.updateState(
- new TaskExecutionStateTransition(
- new TaskExecutionState(
- vertex.getCurrentExecutionAttempt().getAttemptId(),
- state)));
- }
- }
-
- public static void transitionTaskStatus(
- DefaultScheduler scheduler,
- AccessExecutionJobVertex vertex,
- int subtask,
- ExecutionState executionState) {
-
- final ExecutionAttemptID attemptId = vertex.getTaskVertices()[subtask]
- .getCurrentExecutionAttempt()
- .getAttemptId();
- scheduler.updateTaskExecutionState(
- new TaskExecutionState(
- attemptId,
- executionState));
- }
-}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingDownstreamTasksInBatchJobBenchmark.java b/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingDownstreamTasksInBatchJobBenchmarkExecutor.java
similarity index 64%
rename from src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingDownstreamTasksInBatchJobBenchmark.java
rename to src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingDownstreamTasksInBatchJobBenchmarkExecutor.java
index 368263a..4ff2603 100644
--- a/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingDownstreamTasksInBatchJobBenchmark.java
+++ b/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingDownstreamTasksInBatchJobBenchmarkExecutor.java
@@ -19,9 +19,9 @@
package org.apache.flink.scheduler.benchmark.deploying;
import org.apache.flink.runtime.executiongraph.Execution;
-import org.apache.flink.runtime.executiongraph.ExecutionVertex;
-import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.deploying.DeployingDownstreamTasksInBatchJobBenchmark;
+import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -35,39 +35,26 @@ import org.openjdk.jmh.runner.RunnerException;
* The benchmark of deploying downstream tasks in a BATCH job.
* The related method is {@link Execution#deploy}.
*/
-public class DeployingDownstreamTasksInBatchJobBenchmark extends DeployingTasksBenchmarkBase {
+public class DeployingDownstreamTasksInBatchJobBenchmarkExecutor extends SchedulerBenchmarkBase {
@Param("BATCH")
private JobConfiguration jobConfiguration;
- private ExecutionVertex[] vertices;
+ private DeployingDownstreamTasksInBatchJobBenchmark benchmark;
public static void main(String[] args) throws RunnerException {
- runBenchmark(DeployingDownstreamTasksInBatchJobBenchmark.class);
+ runBenchmark(DeployingDownstreamTasksInBatchJobBenchmarkExecutor.class);
}
@Setup(Level.Trial)
public void setup() throws Exception {
- createAndSetupExecutionGraph(jobConfiguration);
-
- final JobVertex source = jobVertices.get(0);
-
- for (ExecutionVertex ev : executionGraph.getJobVertex(source.getID()).getTaskVertices()) {
- Execution execution = ev.getCurrentExecutionAttempt();
- execution.deploy();
- }
-
- final JobVertex sink = jobVertices.get(1);
-
- vertices = executionGraph.getJobVertex(sink.getID()).getTaskVertices();
+ benchmark = new DeployingDownstreamTasksInBatchJobBenchmark();
+ benchmark.setup(jobConfiguration);
}
@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void deployDownstreamTasks() throws Exception {
- for (ExecutionVertex ev : vertices) {
- Execution execution = ev.getCurrentExecutionAttempt();
- execution.deploy();
- }
+ benchmark.deployDownstreamTasks();
}
}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingTasksBenchmarkBase.java b/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingTasksBenchmarkBase.java
deleted file mode 100644
index 601ad09..0000000
--- a/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingTasksBenchmarkBase.java
+++ /dev/null
@@ -1,74 +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.scheduler.benchmark.deploying;
-
-import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
-import org.apache.flink.runtime.executiongraph.Execution;
-import org.apache.flink.runtime.executiongraph.ExecutionGraph;
-import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
-import org.apache.flink.runtime.executiongraph.ExecutionVertex;
-import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway;
-import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.runtime.jobmaster.LogicalSlot;
-import org.apache.flink.runtime.jobmaster.TestingLogicalSlotBuilder;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
-import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
-
-import java.util.List;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.createAndInitExecutionGraph;
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.createDefaultJobVertices;
-
-/**
- * The base class of benchmarks related to deploying tasks.
- */
-public class DeployingTasksBenchmarkBase extends SchedulerBenchmarkBase {
-
- List<JobVertex> jobVertices;
- ExecutionGraph executionGraph;
- BlockingQueue<TaskDeploymentDescriptor> taskDeploymentDescriptors;
-
- public void createAndSetupExecutionGraph(JobConfiguration jobConfiguration) throws Exception {
-
- jobVertices = createDefaultJobVertices(jobConfiguration);
-
- executionGraph = createAndInitExecutionGraph(jobVertices, jobConfiguration);
-
- taskDeploymentDescriptors = new ArrayBlockingQueue<>(jobConfiguration.getParallelism() * 2);
-
- final SimpleAckingTaskManagerGateway taskManagerGateway = new SimpleAckingTaskManagerGateway();
- taskManagerGateway.setSubmitConsumer(taskDeploymentDescriptors::offer);
-
- final TestingLogicalSlotBuilder slotBuilder =
- new TestingLogicalSlotBuilder().setTaskManagerGateway(taskManagerGateway);
-
- for (ExecutionJobVertex ejv : executionGraph.getVerticesTopologically()) {
- for (ExecutionVertex ev : ejv.getTaskVertices()) {
- final LogicalSlot slot = slotBuilder.createTestingLogicalSlot();
- final Execution execution = ev.getCurrentExecutionAttempt();
- execution.registerProducedPartitions(slot.getTaskManagerLocation(), true).get();
- if (!execution.tryAssignResource(slot)) {
- throw new RuntimeException("Error when assigning slot to execution.");
- }
- }
- }
- }
-}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingTasksInStreamingJobBenchmark.java b/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingTasksInStreamingJobBenchmarkExecutor.java
similarity index 75%
rename from src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingTasksInStreamingJobBenchmark.java
rename to src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingTasksInStreamingJobBenchmarkExecutor.java
index 0903089..1c37f86 100644
--- a/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingTasksInStreamingJobBenchmark.java
+++ b/src/main/java/org/apache/flink/scheduler/benchmark/deploying/DeployingTasksInStreamingJobBenchmarkExecutor.java
@@ -19,9 +19,9 @@
package org.apache.flink.scheduler.benchmark.deploying;
import org.apache.flink.runtime.executiongraph.Execution;
-import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
-import org.apache.flink.runtime.executiongraph.ExecutionVertex;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.deploying.DeployingTasksInStreamingJobBenchmark;
+import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -35,28 +35,26 @@ import org.openjdk.jmh.runner.RunnerException;
* The benchmark of deploying tasks in a STREAMING job.
* The related method is {@link Execution#deploy}.
*/
-public class DeployingTasksInStreamingJobBenchmark extends DeployingTasksBenchmarkBase {
+public class DeployingTasksInStreamingJobBenchmarkExecutor extends SchedulerBenchmarkBase {
@Param("STREAMING")
private JobConfiguration jobConfiguration;
+ private DeployingTasksInStreamingJobBenchmark benchmark;
+
public static void main(String[] args) throws RunnerException {
runBenchmark(DeployingTasksInStreamingJobBenchmark.class);
}
@Setup(Level.Trial)
public void setup() throws Exception {
- createAndSetupExecutionGraph(jobConfiguration);
+ benchmark = new DeployingTasksInStreamingJobBenchmark();
+ benchmark.setup(jobConfiguration);
}
@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void deployAllTasks() throws Exception {
- for (ExecutionJobVertex ejv : executionGraph.getVerticesTopologically()) {
- for (ExecutionVertex ev : ejv.getTaskVertices()) {
- Execution execution = ev.getCurrentExecutionAttempt();
- execution.deploy();
- }
- }
+ benchmark.deployAllTasks();
}
}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/failover/FailoverBenchmarkBase.java b/src/main/java/org/apache/flink/scheduler/benchmark/failover/FailoverBenchmarkBase.java
deleted file mode 100644
index 86469e1..0000000
--- a/src/main/java/org/apache/flink/scheduler/benchmark/failover/FailoverBenchmarkBase.java
+++ /dev/null
@@ -1,56 +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.scheduler.benchmark.failover;
-
-import org.apache.flink.runtime.executiongraph.ExecutionGraph;
-import org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy;
-import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
-import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
-import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
-
-import java.util.List;
-import java.util.Set;
-
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.createAndInitExecutionGraph;
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.createDefaultJobVertices;
-
-/**
- * The base class of benchmarks related to failover.
- */
-public class FailoverBenchmarkBase extends SchedulerBenchmarkBase {
-
- JobVertex source;
- List<JobVertex> jobVertices;
-
- ExecutionGraph executionGraph;
- SchedulingTopology schedulingTopology;
- RestartPipelinedRegionFailoverStrategy strategy;
-
- Set<ExecutionVertexID> tasks;
-
- public void createRestartPipelinedRegionFailoverStrategy(JobConfiguration jobConfiguration) throws Exception {
- jobVertices = createDefaultJobVertices(jobConfiguration);
- source = jobVertices.get(0);
- executionGraph = createAndInitExecutionGraph(jobVertices, jobConfiguration);
- schedulingTopology = executionGraph.getSchedulingTopology();
- strategy = new RestartPipelinedRegionFailoverStrategy(schedulingTopology);
- }
-}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInBatchJobBenchmark.java b/src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInBatchJobBenchmarkExecutor.java
similarity index 53%
rename from src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInBatchJobBenchmark.java
rename to src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInBatchJobBenchmarkExecutor.java
index 50becf2..61c0bd4 100644
--- a/src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInBatchJobBenchmark.java
+++ b/src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInBatchJobBenchmarkExecutor.java
@@ -18,11 +18,9 @@
package org.apache.flink.scheduler.benchmark.failover;
-import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy;
-import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.runtime.jobmaster.TestingLogicalSlotBuilder;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.failover.RegionToRestartInBatchJobBenchmark;
+import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -33,44 +31,29 @@ import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.infra.Blackhole;
import org.openjdk.jmh.runner.RunnerException;
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.deployTasks;
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.transitionTaskStatus;
-
/**
* The benchmark of calculating the regions to restart when failover occurs in a BATCH job.
- * The related method is {@link RestartPipelinedRegionFailoverStrategy#getTasksNeedingRestart}.
*/
-public class RegionToRestartInBatchJobBenchmark extends FailoverBenchmarkBase {
+public class RegionToRestartInBatchJobBenchmarkExecutor extends SchedulerBenchmarkBase {
@Param("BATCH")
private JobConfiguration jobConfiguration;
+ private RegionToRestartInBatchJobBenchmark benchmark;
+
public static void main(String[] args) throws RunnerException {
- runBenchmark(RegionToRestartInBatchJobBenchmark.class);
+ runBenchmark(RegionToRestartInBatchJobBenchmarkExecutor.class);
}
@Setup(Level.Trial)
public void setup() throws Exception {
- createRestartPipelinedRegionFailoverStrategy(jobConfiguration);
-
- final JobVertex source = jobVertices.get(0);
- final JobVertex sink = jobVertices.get(1);
-
- final TestingLogicalSlotBuilder slotBuilder = new TestingLogicalSlotBuilder();
-
- deployTasks(executionGraph, source.getID(), slotBuilder, true);
-
- transitionTaskStatus(executionGraph, source.getID(), ExecutionState.FINISHED);
-
- deployTasks(executionGraph, sink.getID(), slotBuilder, true);
+ benchmark = new RegionToRestartInBatchJobBenchmark();
+ benchmark.setup(jobConfiguration);
}
@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void calculateRegionToRestart(Blackhole blackhole) {
- tasks = strategy.getTasksNeedingRestart(
- executionGraph.getJobVertex(source.getID()).getTaskVertices()[0].getID(),
- new Exception("For test."));
- blackhole.consume(tasks);
+ blackhole.consume(benchmark.calculateRegionToRestart());
}
}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInStreamingJobBenchmark.java b/src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInStreamingJobBenchmarkExecutor.java
similarity index 59%
rename from src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInStreamingJobBenchmark.java
rename to src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInStreamingJobBenchmarkExecutor.java
index cb6674a..413e732 100644
--- a/src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInStreamingJobBenchmark.java
+++ b/src/main/java/org/apache/flink/scheduler/benchmark/failover/RegionToRestartInStreamingJobBenchmarkExecutor.java
@@ -18,9 +18,9 @@
package org.apache.flink.scheduler.benchmark.failover;
-import org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy;
-import org.apache.flink.runtime.jobmaster.TestingLogicalSlotBuilder;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.failover.RegionToRestartInStreamingJobBenchmark;
+import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -31,39 +31,29 @@ import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.infra.Blackhole;
import org.openjdk.jmh.runner.RunnerException;
-import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.switchAllVerticesToRunning;
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.deployAllTasks;
-
/**
* The benchmark of calculating region to restart when failover occurs in a STREAMING job.
- * The related method is {@link RestartPipelinedRegionFailoverStrategy#getTasksNeedingRestart}.
*/
-public class RegionToRestartInStreamingJobBenchmark extends FailoverBenchmarkBase {
+public class RegionToRestartInStreamingJobBenchmarkExecutor extends SchedulerBenchmarkBase {
@Param("STREAMING")
private JobConfiguration jobConfiguration;
+ private RegionToRestartInStreamingJobBenchmark benchmark;
+
public static void main(String[] args) throws RunnerException {
- runBenchmark(RegionToRestartInStreamingJobBenchmark.class);
+ runBenchmark(RegionToRestartInStreamingJobBenchmarkExecutor.class);
}
@Setup(Level.Trial)
public void setup() throws Exception {
- createRestartPipelinedRegionFailoverStrategy(jobConfiguration);
-
- TestingLogicalSlotBuilder slotBuilder = new TestingLogicalSlotBuilder();
-
- deployAllTasks(executionGraph, slotBuilder);
-
- switchAllVerticesToRunning(executionGraph);
+ benchmark = new RegionToRestartInStreamingJobBenchmark();
+ benchmark.setup(jobConfiguration);
}
@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void calculateRegionToRestart(Blackhole blackhole) {
- tasks = strategy.getTasksNeedingRestart(
- executionGraph.getJobVertex(source.getID()).getTaskVertices()[0].getID(),
- new Exception("For test."));
- blackhole.consume(tasks);
+ blackhole.consume(benchmark.calculateRegionToRestart());
}
}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java b/src/main/java/org/apache/flink/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java
deleted file mode 100644
index 4426547..0000000
--- a/src/main/java/org/apache/flink/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java
+++ /dev/null
@@ -1,83 +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.scheduler.benchmark.partitionrelease;
-
-import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.executiongraph.ExecutionGraph;
-import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.RegionPartitionReleaseStrategy;
-import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.runtime.jobmaster.TestingLogicalSlotBuilder;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
-import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
-
-import org.openjdk.jmh.annotations.Benchmark;
-import org.openjdk.jmh.annotations.BenchmarkMode;
-import org.openjdk.jmh.annotations.Level;
-import org.openjdk.jmh.annotations.Mode;
-import org.openjdk.jmh.annotations.Param;
-import org.openjdk.jmh.annotations.Setup;
-import org.openjdk.jmh.runner.RunnerException;
-
-import java.util.List;
-
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.createAndInitExecutionGraph;
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.createDefaultJobVertices;
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.deployTasks;
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.transitionTaskStatus;
-
-/**
- * The benchmark of releasing partitions in a BATCH job.
- * The related method is {@link RegionPartitionReleaseStrategy#vertexFinished}.
- */
-public class PartitionReleaseInBatchJobBenchmark extends SchedulerBenchmarkBase {
-
- @Param("BATCH")
- private JobConfiguration jobConfiguration;
-
- private ExecutionGraph executionGraph;
- private JobVertex sink;
-
- public static void main(String[] args) throws RunnerException {
- runBenchmark(PartitionReleaseInBatchJobBenchmark.class);
- }
-
- @Setup(Level.Trial)
- public void setup() throws Exception {
- final List<JobVertex> jobVertices = createDefaultJobVertices(jobConfiguration);
-
- executionGraph = createAndInitExecutionGraph(jobVertices, jobConfiguration);
-
- final JobVertex source = jobVertices.get(0);
- sink = jobVertices.get(1);
-
- final TestingLogicalSlotBuilder slotBuilder = new TestingLogicalSlotBuilder();
-
- deployTasks(executionGraph, source.getID(), slotBuilder, true);
-
- transitionTaskStatus(executionGraph, source.getID(), ExecutionState.FINISHED);
-
- deployTasks(executionGraph, sink.getID(), slotBuilder, true);
- }
-
- @Benchmark
- @BenchmarkMode(Mode.SingleShotTime)
- public void partitionRelease() {
- transitionTaskStatus(executionGraph, sink.getID(), ExecutionState.FINISHED);
- }
-}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmark.java b/src/main/java/org/apache/flink/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmarkExecutor.java
similarity index 62%
copy from src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmark.java
copy to src/main/java/org/apache/flink/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmarkExecutor.java
index 989a6ce..fa1c921 100644
--- a/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmark.java
+++ b/src/main/java/org/apache/flink/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmarkExecutor.java
@@ -16,10 +16,11 @@
* limitations under the License.
*/
-package org.apache.flink.scheduler.benchmark.scheduling;
+package org.apache.flink.scheduler.benchmark.partitionrelease;
-import org.apache.flink.runtime.scheduler.strategy.PipelinedRegionSchedulingStrategy;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.partitionrelease.PartitionReleaseInBatchJobBenchmark;
+import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -27,31 +28,31 @@ import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Setup;
-import org.openjdk.jmh.infra.Blackhole;
import org.openjdk.jmh.runner.RunnerException;
/**
- * The benchmark of initializing {@link PipelinedRegionSchedulingStrategy} in a STREAMING/BATCH job.
+ * The benchmark of releasing partitions in a BATCH job.
*/
-public class InitSchedulingStrategyBenchmark extends SchedulingBenchmarkBase {
+public class PartitionReleaseInBatchJobBenchmarkExecutor extends SchedulerBenchmarkBase {
- @Param({"BATCH", "STREAMING"})
+ @Param("BATCH")
private JobConfiguration jobConfiguration;
+ private PartitionReleaseInBatchJobBenchmark benchmark;
+
public static void main(String[] args) throws RunnerException {
- runBenchmark(InitSchedulingStrategyBenchmark.class);
+ runBenchmark(PartitionReleaseInBatchJobBenchmarkExecutor.class);
}
@Setup(Level.Trial)
public void setup() throws Exception {
- initSchedulingTopology(jobConfiguration);
+ benchmark = new PartitionReleaseInBatchJobBenchmark();
+ benchmark.setup(jobConfiguration);
}
@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
- public void initSchedulingStrategy(Blackhole blackhole) {
- final PipelinedRegionSchedulingStrategy schedulingStrategy =
- new PipelinedRegionSchedulingStrategy(schedulerOperations, schedulingTopology);
- blackhole.consume(schedulingStrategy);
+ public void partitionRelease() {
+ benchmark.partitionRelease();
}
}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmark.java b/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmarkExecutor.java
similarity index 69%
copy from src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmark.java
copy to src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmarkExecutor.java
index 989a6ce..a20e3c7 100644
--- a/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmark.java
+++ b/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmarkExecutor.java
@@ -18,8 +18,9 @@
package org.apache.flink.scheduler.benchmark.scheduling;
-import org.apache.flink.runtime.scheduler.strategy.PipelinedRegionSchedulingStrategy;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.scheduling.InitSchedulingStrategyBenchmark;
+import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -31,27 +32,28 @@ import org.openjdk.jmh.infra.Blackhole;
import org.openjdk.jmh.runner.RunnerException;
/**
- * The benchmark of initializing {@link PipelinedRegionSchedulingStrategy} in a STREAMING/BATCH job.
+ * The benchmark of initializing the scheduling strategy in a STREAMING/BATCH job.
*/
-public class InitSchedulingStrategyBenchmark extends SchedulingBenchmarkBase {
+public class InitSchedulingStrategyBenchmarkExecutor extends SchedulerBenchmarkBase {
@Param({"BATCH", "STREAMING"})
private JobConfiguration jobConfiguration;
+ private InitSchedulingStrategyBenchmark benchmark;
+
public static void main(String[] args) throws RunnerException {
- runBenchmark(InitSchedulingStrategyBenchmark.class);
+ runBenchmark(InitSchedulingStrategyBenchmarkExecutor.class);
}
@Setup(Level.Trial)
public void setup() throws Exception {
- initSchedulingTopology(jobConfiguration);
+ benchmark = new InitSchedulingStrategyBenchmark();
+ benchmark.setup(jobConfiguration);
}
@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void initSchedulingStrategy(Blackhole blackhole) {
- final PipelinedRegionSchedulingStrategy schedulingStrategy =
- new PipelinedRegionSchedulingStrategy(schedulerOperations, schedulingTopology);
- blackhole.consume(schedulingStrategy);
+ blackhole.consume(benchmark.initSchedulingStrategy());
}
}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/SchedulingBenchmarkBase.java b/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/SchedulingBenchmarkBase.java
deleted file mode 100644
index 95a371a..0000000
--- a/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/SchedulingBenchmarkBase.java
+++ /dev/null
@@ -1,49 +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.scheduler.benchmark.scheduling;
-
-import org.apache.flink.runtime.executiongraph.ExecutionGraph;
-import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
-import org.apache.flink.runtime.scheduler.strategy.TestingSchedulerOperations;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
-import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
-
-import java.util.List;
-
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.createAndInitExecutionGraph;
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.createDefaultJobVertices;
-
-/**
- * The base class of benchmarks related to scheduling tasks.
- */
-public class SchedulingBenchmarkBase extends SchedulerBenchmarkBase {
-
- TestingSchedulerOperations schedulerOperations;
- List<JobVertex> jobVertices;
- ExecutionGraph executionGraph;
- SchedulingTopology schedulingTopology;
-
- public void initSchedulingTopology(JobConfiguration jobConfiguration) throws Exception {
- schedulerOperations = new TestingSchedulerOperations();
- jobVertices = createDefaultJobVertices(jobConfiguration);
- executionGraph = createAndInitExecutionGraph(jobVertices, jobConfiguration);
- schedulingTopology = executionGraph.getSchedulingTopology();
- }
-}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/SchedulingDownstreamTasksInBatchJobBenchmark.java b/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/SchedulingDownstreamTasksInBatchJobBenchmark.java
deleted file mode 100644
index 31f752c..0000000
--- a/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/SchedulingDownstreamTasksInBatchJobBenchmark.java
+++ /dev/null
@@ -1,80 +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.scheduler.benchmark.scheduling;
-
-import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.executiongraph.IntermediateResult;
-import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
-import org.apache.flink.runtime.scheduler.strategy.PipelinedRegionSchedulingStrategy;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
-
-import org.openjdk.jmh.annotations.Benchmark;
-import org.openjdk.jmh.annotations.BenchmarkMode;
-import org.openjdk.jmh.annotations.Level;
-import org.openjdk.jmh.annotations.Mode;
-import org.openjdk.jmh.annotations.Param;
-import org.openjdk.jmh.annotations.Setup;
-import org.openjdk.jmh.runner.RunnerException;
-
-import java.lang.reflect.Field;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * The benchmark of scheduling downstream task in a BATCH job.
- * The related method is {@link PipelinedRegionSchedulingStrategy#onExecutionStateChange}.
- */
-public class SchedulingDownstreamTasksInBatchJobBenchmark extends SchedulingBenchmarkBase {
-
- @Param({"BATCH"})
- private JobConfiguration jobConfiguration;
-
- private ExecutionVertexID executionVertexID;
- private PipelinedRegionSchedulingStrategy schedulingStrategy;
-
- public static void main(String[] args) throws RunnerException {
- runBenchmark(SchedulingDownstreamTasksInBatchJobBenchmark.class);
- }
-
- @Setup(Level.Trial)
- public void setup() throws Exception {
- initSchedulingTopology(jobConfiguration);
-
- schedulingStrategy = new PipelinedRegionSchedulingStrategy(schedulerOperations, schedulingTopology);
-
- // When we trying to scheduling downstream tasks via onExecutionStateChange(ExecutionState.FINISHED),
- // the result partitions of upstream tasks need to be CONSUMABLE.
- // The CONSUMABLE status is determined by the variable "numberOfRunningProducers" of the IntermediateResult.
- // Its value cannot be changed by any public methods.
- // So here we use reflections to modify this value and then schedule the downstream tasks.
- for (IntermediateResult result : executionGraph.getAllIntermediateResults().values()) {
- Field f = result.getClass().getDeclaredField("numberOfRunningProducers");
- f.setAccessible(true);
- AtomicInteger numberOfRunningProducers = (AtomicInteger) f.get(result);
- numberOfRunningProducers.set(0);
- }
-
- executionVertexID = executionGraph.getJobVertex(jobVertices.get(0).getID()).getTaskVertices()[0].getID();
- }
-
- @Benchmark
- @BenchmarkMode(Mode.SingleShotTime)
- public void schedulingDownstreamTasks() {
- schedulingStrategy.onExecutionStateChange(executionVertexID, ExecutionState.FINISHED);
- }
-}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmark.java b/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/SchedulingDownstreamTasksInBatchJobBenchmarkExecutor.java
similarity index 65%
rename from src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmark.java
rename to src/main/java/org/apache/flink/scheduler/benchmark/scheduling/SchedulingDownstreamTasksInBatchJobBenchmarkExecutor.java
index 989a6ce..c6de8fe 100644
--- a/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/InitSchedulingStrategyBenchmark.java
+++ b/src/main/java/org/apache/flink/scheduler/benchmark/scheduling/SchedulingDownstreamTasksInBatchJobBenchmarkExecutor.java
@@ -18,8 +18,9 @@
package org.apache.flink.scheduler.benchmark.scheduling;
-import org.apache.flink.runtime.scheduler.strategy.PipelinedRegionSchedulingStrategy;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.scheduling.SchedulingDownstreamTasksInBatchJobBenchmark;
+import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -27,31 +28,31 @@ import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Setup;
-import org.openjdk.jmh.infra.Blackhole;
import org.openjdk.jmh.runner.RunnerException;
/**
- * The benchmark of initializing {@link PipelinedRegionSchedulingStrategy} in a STREAMING/BATCH job.
+ * The benchmark of scheduling downstream task in a BATCH job.
*/
-public class InitSchedulingStrategyBenchmark extends SchedulingBenchmarkBase {
+public class SchedulingDownstreamTasksInBatchJobBenchmarkExecutor extends SchedulerBenchmarkBase {
- @Param({"BATCH", "STREAMING"})
+ @Param({"BATCH"})
private JobConfiguration jobConfiguration;
+ private SchedulingDownstreamTasksInBatchJobBenchmark benchmark;
+
public static void main(String[] args) throws RunnerException {
- runBenchmark(InitSchedulingStrategyBenchmark.class);
+ runBenchmark(SchedulingDownstreamTasksInBatchJobBenchmarkExecutor.class);
}
@Setup(Level.Trial)
public void setup() throws Exception {
- initSchedulingTopology(jobConfiguration);
+ benchmark = new SchedulingDownstreamTasksInBatchJobBenchmark();
+ benchmark.setup(jobConfiguration);
}
@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
- public void initSchedulingStrategy(Blackhole blackhole) {
- final PipelinedRegionSchedulingStrategy schedulingStrategy =
- new PipelinedRegionSchedulingStrategy(schedulerOperations, schedulingTopology);
- blackhole.consume(schedulingStrategy);
+ public void schedulingDownstreamTasks() {
+ benchmark.schedulingDownstreamTasks();
}
}
diff --git a/src/main/java/org/apache/flink/scheduler/benchmark/topology/BuildExecutionGraphBenchmark.java b/src/main/java/org/apache/flink/scheduler/benchmark/topology/BuildExecutionGraphBenchmarkExecutor.java
similarity index 57%
rename from src/main/java/org/apache/flink/scheduler/benchmark/topology/BuildExecutionGraphBenchmark.java
rename to src/main/java/org/apache/flink/scheduler/benchmark/topology/BuildExecutionGraphBenchmarkExecutor.java
index acce988..586f535 100644
--- a/src/main/java/org/apache/flink/scheduler/benchmark/topology/BuildExecutionGraphBenchmark.java
+++ b/src/main/java/org/apache/flink/scheduler/benchmark/topology/BuildExecutionGraphBenchmarkExecutor.java
@@ -18,11 +18,8 @@
package org.apache.flink.scheduler.benchmark.topology;
-import org.apache.flink.runtime.executiongraph.ExecutionGraph;
-import org.apache.flink.runtime.executiongraph.TestingDefaultExecutionGraphBuilder;
-import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.jobgraph.JobVertex;
-import org.apache.flink.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.JobConfiguration;
+import org.apache.flink.runtime.scheduler.benchmark.topology.BuildExecutionGraphBenchmark;
import org.apache.flink.scheduler.benchmark.SchedulerBenchmarkBase;
import org.openjdk.jmh.annotations.Benchmark;
@@ -33,37 +30,29 @@ import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.runner.RunnerException;
-import java.util.List;
-
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.createDefaultJobVertices;
-import static org.apache.flink.scheduler.benchmark.SchedulerBenchmarkUtils.createJobGraph;
-
/**
- * The benchmark of building the topology of {@link ExecutionGraph} in a STREAMING/BATCH job.
- * The related method is {@link ExecutionGraph#attachJobGraph},
+ * The benchmark of building the topology of ExecutionGraph in a STREAMING/BATCH job.
*/
-public class BuildExecutionGraphBenchmark extends SchedulerBenchmarkBase {
-
- private List<JobVertex> jobVertices;
- private ExecutionGraph executionGraph;
+public class BuildExecutionGraphBenchmarkExecutor extends SchedulerBenchmarkBase {
@Param({"BATCH", "STREAMING"})
private JobConfiguration jobConfiguration;
+ private BuildExecutionGraphBenchmark benchmark;
+
public static void main(String[] args) throws RunnerException {
- runBenchmark(BuildExecutionGraphBenchmark.class);
+ runBenchmark(BuildExecutionGraphBenchmarkExecutor.class);
}
@Setup(Level.Trial)
public void setup() throws Exception {
- jobVertices = createDefaultJobVertices(jobConfiguration);
- final JobGraph jobGraph = createJobGraph(jobConfiguration);
- executionGraph = TestingDefaultExecutionGraphBuilder.newBuilder().setJobGraph(jobGraph).build();
+ benchmark = new BuildExecutionGraphBenchmark();
+ benchmark.setup(jobConfiguration);
}
@Benchmark
@BenchmarkMode(Mode.SingleShotTime)
public void buildTopology() throws Exception {
- executionGraph.attachJobGraph(jobVertices);
+ benchmark.buildTopology();
}
}