You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2019/12/07 15:33:20 UTC
[flink] branch release-1.9 updated: [FLINK-14735][scheduler]
Improve scheduling of all-to-all partitions with ALL input constraint for
legacy scheduler
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.9 by this push:
new bfb4e41 [FLINK-14735][scheduler] Improve scheduling of all-to-all partitions with ALL input constraint for legacy scheduler
bfb4e41 is described below
commit bfb4e415363526e87d606d9ef73da8d98b23ad6e
Author: Zhu Zhu <re...@gmail.com>
AuthorDate: Fri Nov 15 14:58:31 2019 +0800
[FLINK-14735][scheduler] Improve scheduling of all-to-all partitions with ALL input constraint for legacy scheduler
This closes #10283
---
.../flink/runtime/executiongraph/Execution.java | 58 ++++++++++++-------
.../ExecutionVertexInputConstraintTest.java | 65 ++++++++++++++++++++--
2 files changed, 97 insertions(+), 26 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index c070e4e..1404aa1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -73,6 +73,7 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -860,10 +861,19 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
void scheduleOrUpdateConsumers(List<List<ExecutionEdge>> allConsumers) {
assertRunningInJobMasterMainThread();
- final int numConsumers = allConsumers.size();
- if (numConsumers > 1) {
+ final HashSet<ExecutionVertex> consumerDeduplicator = new HashSet<>();
+ scheduleOrUpdateConsumers(allConsumers, consumerDeduplicator);
+ }
+
+ private void scheduleOrUpdateConsumers(
+ final List<List<ExecutionEdge>> allConsumers,
+ final HashSet<ExecutionVertex> consumerDeduplicator) {
+
+ if (allConsumers.size() == 0) {
+ return;
+ }
+ if (allConsumers.size() > 1) {
fail(new IllegalStateException("Currently, only a single consumer group per partition is supported."));
- } else if (numConsumers == 0) {
return;
}
@@ -873,8 +883,7 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
final ExecutionState consumerState = consumer.getState();
// ----------------------------------------------------------------
- // Consumer is created => try to schedule it and the partition info
- // is known during deployment
+ // Consumer is created => needs to be scheduled
// ----------------------------------------------------------------
if (consumerState == CREATED) {
// Schedule the consumer vertex if its inputs constraint is satisfied, otherwise skip the scheduling.
@@ -882,8 +891,11 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
// at least one of the consumer vertex's inputs is consumable here. This is to avoid the
// O(N) complexity introduced by input constraint check for InputDependencyConstraint.ANY,
// as we do not want the default scheduling performance to be affected.
- if (consumerVertex.getInputDependencyConstraint() == InputDependencyConstraint.ANY ||
- consumerVertex.checkInputDependencyConstraints()) {
+
+ if (consumerDeduplicator.add(consumerVertex) &&
+ (consumerVertex.getInputDependencyConstraint() == InputDependencyConstraint.ANY ||
+ consumerVertex.checkInputDependencyConstraints())) {
+
scheduleConsumer(consumerVertex);
}
}
@@ -1054,21 +1066,9 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
if (transitionState(current, FINISHED)) {
try {
- for (IntermediateResultPartition finishedPartition
- : getVertex().finishAllBlockingPartitions()) {
-
- IntermediateResultPartition[] allPartitions = finishedPartition
- .getIntermediateResult().getPartitions();
-
- for (IntermediateResultPartition partition : allPartitions) {
- scheduleOrUpdateConsumers(partition.getConsumers());
- }
- }
-
+ finishPartitionsAndScheduleOrUpdateConsumers();
updateAccumulatorsAndMetrics(userAccumulators, metrics);
-
releaseAssignedResource(null);
-
vertex.getExecutionGraph().deregisterExecution(this);
}
finally {
@@ -1097,6 +1097,24 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
}
}
+ private void finishPartitionsAndScheduleOrUpdateConsumers() {
+ final List<IntermediateResultPartition> newlyFinishedResults = getVertex().finishAllBlockingPartitions();
+ if (newlyFinishedResults.isEmpty()) {
+ return;
+ }
+
+ final HashSet<ExecutionVertex> consumerDeduplicator = new HashSet<>();
+
+ for (IntermediateResultPartition finishedPartition : newlyFinishedResults) {
+ final IntermediateResultPartition[] allPartitionsOfNewlyFinishedResults =
+ finishedPartition.getIntermediateResult().getPartitions();
+
+ for (IntermediateResultPartition partition : allPartitionsOfNewlyFinishedResults) {
+ scheduleOrUpdateConsumers(partition.getConsumers(), consumerDeduplicator);
+ }
+ }
+ }
+
private boolean cancelAtomically() {
if (startCancelling(0)) {
completeCancelling();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexInputConstraintTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexInputConstraintTest.java
index 9aaed7c..6df210b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexInputConstraintTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexInputConstraintTest.java
@@ -31,13 +31,16 @@ import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobmaster.slotpool.SlotProvider;
+import org.apache.flink.runtime.taskmanager.TaskExecutionState;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
+import java.time.Duration;
import java.util.Arrays;
import java.util.List;
@@ -45,7 +48,9 @@ import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.is
import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.waitForAllExecutionsPredicate;
import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.waitUntilExecutionVertexState;
import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.waitUntilJobStatus;
+import static org.hamcrest.Matchers.lessThan;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
/**
@@ -109,7 +114,6 @@ public class ExecutionVertexInputConstraintTest extends TestLogger {
eg.start(mainThreadExecutor);
eg.scheduleForExecution();
-
// Inputs constraint not satisfied on init
assertFalse(ev31.checkInputDependencyConstraints());
@@ -122,10 +126,8 @@ public class ExecutionVertexInputConstraintTest extends TestLogger {
// Inputs constraint not satisfied after failover
ev11.fail(new Exception());
-
waitUntilJobRestarted(eg);
-
assertFalse(ev31.checkInputDependencyConstraints());
// Input2 consumable satisfies the constraint
@@ -149,7 +151,6 @@ public class ExecutionVertexInputConstraintTest extends TestLogger {
eg.start(mainThreadExecutor);
eg.scheduleForExecution();
-
// Inputs constraint not satisfied on init
assertFalse(ev31.checkInputDependencyConstraints());
@@ -167,12 +168,56 @@ public class ExecutionVertexInputConstraintTest extends TestLogger {
// Inputs constraint not satisfied after failover
ev11.fail(new Exception());
-
waitUntilJobRestarted(eg);
assertFalse(ev31.checkInputDependencyConstraints());
}
+ @Test
+ public void testInputConstraintALLPerformance() throws Exception {
+ final int parallelism = 1000;
+ final JobVertex v1 = createVertexWithAllInputConstraints("vertex1", parallelism);
+ final JobVertex v2 = createVertexWithAllInputConstraints("vertex2", parallelism);
+ final JobVertex v3 = createVertexWithAllInputConstraints("vertex3", parallelism);
+ v2.connectNewDataSetAsInput(v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.BLOCKING);
+ v2.connectNewDataSetAsInput(v3, DistributionPattern.ALL_TO_ALL, ResultPartitionType.BLOCKING);
+
+ final ExecutionGraph eg = createExecutionGraph(Arrays.asList(v1, v3, v2), InputDependencyConstraint.ALL, 3000);
+
+ eg.start(mainThreadExecutor);
+ eg.scheduleForExecution();
+
+ for (int i = 0; i < parallelism - 1; i++) {
+ finishSubtask(eg, v1.getID(), i);
+ }
+
+ final long startTime = System.nanoTime();
+ finishSubtask(eg, v1.getID(), parallelism - 1);
+
+ final Duration duration = Duration.ofNanos(System.nanoTime() - startTime);
+ final Duration timeout = Duration.ofSeconds(5);
+
+ assertThat(duration, lessThan(timeout));
+ }
+
+ private static JobVertex createVertexWithAllInputConstraints(String name, int parallelism) {
+ final JobVertex v = new JobVertex(name);
+ v.setParallelism(parallelism);
+ v.setInvokableClass(AbstractInvokable.class);
+ v.setInputDependencyConstraint(InputDependencyConstraint.ALL);
+ return v;
+ }
+
+ private static void finishSubtask(ExecutionGraph graph, JobVertexID jvId, int subtask) {
+ final ExecutionVertex[] vertices = graph.getJobVertex(jvId).getTaskVertices();
+
+ graph.updateState(
+ new TaskExecutionState(
+ graph.getJobID(),
+ vertices[subtask].getCurrentExecutionAttempt().getAttemptId(),
+ ExecutionState.FINISHED));
+ }
+
private static List<JobVertex> createOrderedVertices() {
JobVertex v1 = new JobVertex("vertex1");
JobVertex v2 = new JobVertex("vertex2");
@@ -192,9 +237,17 @@ public class ExecutionVertexInputConstraintTest extends TestLogger {
List<JobVertex> orderedVertices,
InputDependencyConstraint inputDependencyConstraint) throws Exception {
+ return createExecutionGraph(orderedVertices, inputDependencyConstraint, 20);
+ }
+
+ private static ExecutionGraph createExecutionGraph(
+ List<JobVertex> orderedVertices,
+ InputDependencyConstraint inputDependencyConstraint,
+ int numSlots) throws Exception {
+
final JobID jobId = new JobID();
final String jobName = "Test Job Sample Name";
- final SlotProvider slotProvider = new SimpleSlotProvider(jobId, 20);
+ final SlotProvider slotProvider = new SimpleSlotProvider(jobId, numSlots);
for (JobVertex vertex : orderedVertices) {
vertex.setInputDependencyConstraint(inputDependencyConstraint);