[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[
https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16717364#comment-16717364
]
ASF GitHub Bot commented on FLINK-10945:
zhuzhurk commented on a change in pull request #7255: [FLINK-10945] Use
InputDependencyConstraint to avoid resource dead…
URL: https://github.com/apache/flink/pull/7255#discussion_r240655589
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
##
@@ -726,6 +728,56 @@ void sendPartitionInfos() {
}
}
+ /**
+* Check whether the InputDependencyConstraint is satisfied for this
vertex.
+*
+* @return whether the input constraint is satisfied
+*/
+ public boolean checkInputDependencyConstraints() {
Review comment:
Thanks for the suggestion. I made it more concise, but a bit different from
your sample code:
For PIPELINED input consumable check, only the partitions related with the
input edges are checked.
I'm not sure why you think the behavior is changed?
I'd agree if you have concerns that this check could be heavy(O(N)
complexity)) for large parallelism vertex. So I'm considering whether to add a
short cut check for `InputDependencyConstraint.ANY` in
`Execution.scheduleOrUpdateConsumers`, i.e. change `if
(consumerVertex.checkInputDependencyConstraints()) ` to `if
(getVertex().getExecutionGraph().getInputDependencyConstraint() ==
InputDependencyConstraint.ANY ||
consumerVertex.checkInputDependencyConstraints())`. It's not elegant but can be
as efficient as current scheduling.
The input state is not used in EAGER scheduling. So the
InputDependencyConstraint is not a matter in EAGER mode. It is used in
scheduleOrUpdateConsumers, which is only invoked in LAZY_FROM_SOURCES
scheduling.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Avoid resource deadlocks for finite stream jobs when resources are limited
> --
>
> Key: FLINK-10945
> URL: https://issues.apache.org/jira/browse/FLINK-10945
> Project: Flink
> Issue Type: Improvement
> Components: Distributed Coordination
>Affects Versions: 1.7.1
>Reporter: Zhu Zhu
>Assignee: Zhu Zhu
>Priority: Major
> Labels: pull-request-available
>
> Currently *resource deadlocks* can happen to finite stream jobs(or batch
> jobs) when resources are limited. In 2 cases as below:
> # Task Y is a pipelined downstream task of task X. When X takes all
> resources(slots), Y cannot acquire slots to start, thus the back pressure
> will block X to finish
> # Task Y is a upstream task of task X. When X takes all resources(slots) and
> Y cannot start, X cannot finish as some of its inputs are not finished
>
> We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline
> back pressure. However, case 2 cannot be avoided as X(downstream task) will
> be launched when any of its input result is ready.
> To be detailed, say task X has BLOCKING upstream task Y and Z, X can be
> launched when Z finishes, though task Y is not launched yet. This pre-launch
> behaviour can be beneficial when there are plenty of resources, thus X can
> process data from Z earlier before Y finishes its data processing. However,
> resource deadlocks may happen when the resources are limited, e.g. in small
> sessions.
>
> I’d propose introducing a constraint named as *InputDependencyConstraint* to
> control the scheduling of vertices. It has 2 values:
> # *ANY*. The vertex can be scheduled when any of its inputs is consumable.
> # *ALL*. The vertex can be scheduled when all of its inputs are consumable.
>
> The design doc is here.
> [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[
https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16717254#comment-16717254
]
ASF GitHub Bot commented on FLINK-10945:
zhuzhurk commented on a change in pull request #7255: [FLINK-10945] Use
InputDependencyConstraint to avoid resource dead…
URL: https://github.com/apache/flink/pull/7255#discussion_r240629262
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
##
@@ -747,31 +747,34 @@ else if (numConsumers == 0) {
consumerVertex.cachePartitionInfo(PartialInputChannelDeploymentDescriptor.fromEdge(
partition, partitionExecution));
- // When deploying a consuming task, its task
deployment descriptor will contain all
- // deployment information available at the
respective time. It is possible that some
- // of the partitions to be consumed have not
been created yet. These are updated
- // runtime via the update messages.
- //
- // TODO The current approach may send many
update messages even though the consuming
- // task has already been deployed with all
necessary information. We have to check
- // whether this is a problem and fix it, if it
is.
- CompletableFuture.supplyAsync(
- () -> {
- try {
- final ExecutionGraph
executionGraph = consumerVertex.getExecutionGraph();
-
consumerVertex.scheduleForExecution(
-
executionGraph.getSlotProvider(),
-
executionGraph.isQueuedSchedulingAllowed(),
-
LocationPreferenceConstraint.ANY, // there must be at least one known location
-
Collections.emptySet());
- } catch (Throwable t) {
- consumerVertex.fail(new
IllegalStateException("Could not schedule consumer " +
+ // Schedule the consumer vertex if its inputs
constraint is satisfied, otherwise postpone the scheduling
+ if
(consumerVertex.checkInputDependencyConstraints()) {
+ // When deploying a consuming task, its
task deployment descriptor will contain all
+ // deployment information available at
the respective time. It is possible that some
+ // of the partitions to be consumed
have not been created yet. These are updated
+ // runtime via the update messages.
+ //
+ // TODO The current approach may send
many update messages even though the consuming
+ // task has already been deployed with
all necessary information. We have to check
+ // whether this is a problem and fix
it, if it is.
+ CompletableFuture.supplyAsync(
Review comment:
Sure.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Avoid resource deadlocks for finite stream jobs when resources are limited
> --
>
> Key: FLINK-10945
> URL: https://issues.apache.org/jira/browse/FLINK-10945
> Project: Flink
> Issue Type: Improvement
> Components: Distributed Coordination
>Affects Versions: 1.7.1
>Reporter: Zhu Zhu
>Assignee: Zhu Zhu
>Priority: Major
> Labels: pull-request-available
>
> Currently *resource deadlocks* can happen to finite stream jobs(or batch
> jobs) when resources are limited. In 2 cases as below:
> # Task Y is a pipelined downstream task of task X. When X takes all
> resources(slots), Y cannot acquire slots to start, thus the back pressure
> will block X to finish
> # Task Y is a upstream task of task X. When
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16717255#comment-16717255 ] ASF GitHub Bot commented on FLINK-10945: zhuzhurk commented on a change in pull request #7255: [FLINK-10945] Use InputDependencyConstraint to avoid resource dead… URL: https://github.com/apache/flink/pull/7255#discussion_r240629621 ## File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java ## @@ -36,6 +36,11 @@ private List> consumers; + /** +* Whether this partition has data produced. For pipelined result only. Review comment: Agree. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > Labels: pull-request-available > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > > The design doc is here. > [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16717252#comment-16717252 ] ASF GitHub Bot commented on FLINK-10945: zhuzhurk commented on a change in pull request #7255: [FLINK-10945] Use InputDependencyConstraint to avoid resource dead… URL: https://github.com/apache/flink/pull/7255#discussion_r240629143 ## File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java ## @@ -36,6 +36,11 @@ private List> consumers; + /** +* Whether this partition has data produced. For pipelined result only. +*/ + private boolean dataProduced = false; Review comment: That's fine. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > Labels: pull-request-available > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > > The design doc is here. > [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[
https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16717251#comment-16717251
]
ASF GitHub Bot commented on FLINK-10945:
zhuzhurk commented on a change in pull request #7255: [FLINK-10945] Use
InputDependencyConstraint to avoid resource dead…
URL: https://github.com/apache/flink/pull/7255#discussion_r240628670
##
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptorTest.java
##
@@ -192,6 +193,7 @@ private static ExecutionVertex
mockExecutionVertex(ExecutionState state, Resourc
private static IntermediateResultPartition
mockPartition(ExecutionVertex producer) {
IntermediateResultPartition partition =
mock(IntermediateResultPartition.class);
+
when(partition.getResultType()).thenReturn(ResultPartitionType.PIPELINED);
Review comment:
Yes. It is needed to avoid NPE in the tests.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Avoid resource deadlocks for finite stream jobs when resources are limited
> --
>
> Key: FLINK-10945
> URL: https://issues.apache.org/jira/browse/FLINK-10945
> Project: Flink
> Issue Type: Improvement
> Components: Distributed Coordination
>Affects Versions: 1.7.1
>Reporter: Zhu Zhu
>Assignee: Zhu Zhu
>Priority: Major
> Labels: pull-request-available
>
> Currently *resource deadlocks* can happen to finite stream jobs(or batch
> jobs) when resources are limited. In 2 cases as below:
> # Task Y is a pipelined downstream task of task X. When X takes all
> resources(slots), Y cannot acquire slots to start, thus the back pressure
> will block X to finish
> # Task Y is a upstream task of task X. When X takes all resources(slots) and
> Y cannot start, X cannot finish as some of its inputs are not finished
>
> We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline
> back pressure. However, case 2 cannot be avoided as X(downstream task) will
> be launched when any of its input result is ready.
> To be detailed, say task X has BLOCKING upstream task Y and Z, X can be
> launched when Z finishes, though task Y is not launched yet. This pre-launch
> behaviour can be beneficial when there are plenty of resources, thus X can
> process data from Z earlier before Y finishes its data processing. However,
> resource deadlocks may happen when the resources are limited, e.g. in small
> sessions.
>
> I’d propose introducing a constraint named as *InputDependencyConstraint* to
> control the scheduling of vertices. It has 2 values:
> # *ANY*. The vertex can be scheduled when any of its inputs is consumable.
> # *ALL*. The vertex can be scheduled when all of its inputs are consumable.
>
> The design doc is here.
> [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[
https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16715279#comment-16715279
]
ASF GitHub Bot commented on FLINK-10945:
azagrebin commented on a change in pull request #7255: [FLINK-10945] Use
InputDependencyConstraint to avoid resource dead…
URL: https://github.com/apache/flink/pull/7255#discussion_r240313527
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
##
@@ -726,6 +728,56 @@ void sendPartitionInfos() {
}
}
+ /**
+* Check whether the InputDependencyConstraint is satisfied for this
vertex.
+*
+* @return whether the input constraint is satisfied
+*/
+ public boolean checkInputDependencyConstraints() {
Review comment:
This might be a bit more concise:
```
boolean checkInputDependencyConstraints() {
if (getExecutionGraph().getInputDependencyConstraint() ==
InputDependencyConstraint.ANY) {
// InputDependencyConstraint == ANY
return
jobVertex.getInputs().stream().anyMatch(this::isInputConsumable);
} else {
// InputDependencyConstraint == ALL
return
jobVertex.getInputs().stream().allMatch(this::isInputConsumable);
}
}
boolean isInputConsumable(IntermediateResult result) {
if (result.getResultType().isPipelined()) {
// For PIPELINED result, the input is consumable if any
result partition has produced records or is finished
return
Arrays.stream(result.getPartitions()).anyMatch(IntermediateResultPartition::hasDataProduced);
} else {
// For BLOCKING result, the input is consumable if all
the partitions in the result are finished
return result.areAllPartitionsFinished();
}
}
```
I am not sure we need to check the `ANY` case at all. Just checking this
theoretically changes current behaviour. On the other hand, at the moment, I
think it is always true for `ANY` where we check it if
`ScheduleMode.LAZY_FROM_SOURCES`.
I am also not sure that `ALL` config makes sense together with
`ScheduleMode.EAGER`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Avoid resource deadlocks for finite stream jobs when resources are limited
> --
>
> Key: FLINK-10945
> URL: https://issues.apache.org/jira/browse/FLINK-10945
> Project: Flink
> Issue Type: Improvement
> Components: Distributed Coordination
>Affects Versions: 1.7.1
>Reporter: Zhu Zhu
>Assignee: Zhu Zhu
>Priority: Major
> Labels: pull-request-available
>
> Currently *resource deadlocks* can happen to finite stream jobs(or batch
> jobs) when resources are limited. In 2 cases as below:
> # Task Y is a pipelined downstream task of task X. When X takes all
> resources(slots), Y cannot acquire slots to start, thus the back pressure
> will block X to finish
> # Task Y is a upstream task of task X. When X takes all resources(slots) and
> Y cannot start, X cannot finish as some of its inputs are not finished
>
> We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline
> back pressure. However, case 2 cannot be avoided as X(downstream task) will
> be launched when any of its input result is ready.
> To be detailed, say task X has BLOCKING upstream task Y and Z, X can be
> launched when Z finishes, though task Y is not launched yet. This pre-launch
> behaviour can be beneficial when there are plenty of resources, thus X can
> process data from Z earlier before Y finishes its data processing. However,
> resource deadlocks may happen when the resources are limited, e.g. in small
> sessions.
>
> I’d propose introducing a constraint named as *InputDependencyConstraint* to
> control the scheduling of vertices. It has 2 values:
> # *ANY*. The vertex can be scheduled when any of its inputs is consumable.
> # *ALL*. The vertex can be scheduled when all of its inputs are consumable.
>
> The design doc is here.
> [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16715278#comment-16715278 ] ASF GitHub Bot commented on FLINK-10945: azagrebin commented on a change in pull request #7255: [FLINK-10945] Use InputDependencyConstraint to avoid resource dead… URL: https://github.com/apache/flink/pull/7255#discussion_r240312767 ## File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java ## @@ -36,6 +36,11 @@ private List> consumers; + /** +* Whether this partition has data produced. For pipelined result only. Review comment: Whether this partition has produced some data. For pipelined result only. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > Labels: pull-request-available > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > > The design doc is here. > [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16715276#comment-16715276 ] ASF GitHub Bot commented on FLINK-10945: azagrebin commented on a change in pull request #7255: [FLINK-10945] Use InputDependencyConstraint to avoid resource dead… URL: https://github.com/apache/flink/pull/7255#discussion_r240311956 ## File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java ## @@ -36,6 +36,11 @@ private List> consumers; + /** +* Whether this partition has data produced. For pipelined result only. +*/ + private boolean dataProduced = false; Review comment: I would call it `isSomePipelinedDataProduced`. Also methid: `markSomePipelinedDataProduced` This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > Labels: pull-request-available > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > > The design doc is here. > [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[
https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16715275#comment-16715275
]
ASF GitHub Bot commented on FLINK-10945:
azagrebin commented on a change in pull request #7255: [FLINK-10945] Use
InputDependencyConstraint to avoid resource dead…
URL: https://github.com/apache/flink/pull/7255#discussion_r240321678
##
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptorTest.java
##
@@ -192,6 +193,7 @@ private static ExecutionVertex
mockExecutionVertex(ExecutionState state, Resourc
private static IntermediateResultPartition
mockPartition(ExecutionVertex producer) {
IntermediateResultPartition partition =
mock(IntermediateResultPartition.class);
+
when(partition.getResultType()).thenReturn(ResultPartitionType.PIPELINED);
Review comment:
Is this change needed? `isConsumable` should be still enough.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Avoid resource deadlocks for finite stream jobs when resources are limited
> --
>
> Key: FLINK-10945
> URL: https://issues.apache.org/jira/browse/FLINK-10945
> Project: Flink
> Issue Type: Improvement
> Components: Distributed Coordination
>Affects Versions: 1.7.1
>Reporter: Zhu Zhu
>Assignee: Zhu Zhu
>Priority: Major
> Labels: pull-request-available
>
> Currently *resource deadlocks* can happen to finite stream jobs(or batch
> jobs) when resources are limited. In 2 cases as below:
> # Task Y is a pipelined downstream task of task X. When X takes all
> resources(slots), Y cannot acquire slots to start, thus the back pressure
> will block X to finish
> # Task Y is a upstream task of task X. When X takes all resources(slots) and
> Y cannot start, X cannot finish as some of its inputs are not finished
>
> We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline
> back pressure. However, case 2 cannot be avoided as X(downstream task) will
> be launched when any of its input result is ready.
> To be detailed, say task X has BLOCKING upstream task Y and Z, X can be
> launched when Z finishes, though task Y is not launched yet. This pre-launch
> behaviour can be beneficial when there are plenty of resources, thus X can
> process data from Z earlier before Y finishes its data processing. However,
> resource deadlocks may happen when the resources are limited, e.g. in small
> sessions.
>
> I’d propose introducing a constraint named as *InputDependencyConstraint* to
> control the scheduling of vertices. It has 2 values:
> # *ANY*. The vertex can be scheduled when any of its inputs is consumable.
> # *ALL*. The vertex can be scheduled when all of its inputs are consumable.
>
> The design doc is here.
> [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[
https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16715277#comment-16715277
]
ASF GitHub Bot commented on FLINK-10945:
azagrebin commented on a change in pull request #7255: [FLINK-10945] Use
InputDependencyConstraint to avoid resource dead…
URL: https://github.com/apache/flink/pull/7255#discussion_r240310923
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
##
@@ -747,31 +747,34 @@ else if (numConsumers == 0) {
consumerVertex.cachePartitionInfo(PartialInputChannelDeploymentDescriptor.fromEdge(
partition, partitionExecution));
- // When deploying a consuming task, its task
deployment descriptor will contain all
- // deployment information available at the
respective time. It is possible that some
- // of the partitions to be consumed have not
been created yet. These are updated
- // runtime via the update messages.
- //
- // TODO The current approach may send many
update messages even though the consuming
- // task has already been deployed with all
necessary information. We have to check
- // whether this is a problem and fix it, if it
is.
- CompletableFuture.supplyAsync(
- () -> {
- try {
- final ExecutionGraph
executionGraph = consumerVertex.getExecutionGraph();
-
consumerVertex.scheduleForExecution(
-
executionGraph.getSlotProvider(),
-
executionGraph.isQueuedSchedulingAllowed(),
-
LocationPreferenceConstraint.ANY, // there must be at least one known location
-
Collections.emptySet());
- } catch (Throwable t) {
- consumerVertex.fail(new
IllegalStateException("Could not schedule consumer " +
+ // Schedule the consumer vertex if its inputs
constraint is satisfied, otherwise postpone the scheduling
+ if
(consumerVertex.checkInputDependencyConstraints()) {
+ // When deploying a consuming task, its
task deployment descriptor will contain all
+ // deployment information available at
the respective time. It is possible that some
+ // of the partitions to be consumed
have not been created yet. These are updated
+ // runtime via the update messages.
+ //
+ // TODO The current approach may send
many update messages even though the consuming
+ // task has already been deployed with
all necessary information. We have to check
+ // whether this is a problem and fix
it, if it is.
+ CompletableFuture.supplyAsync(
Review comment:
The body of the introduced if is quite big, could we move it into a separate
method, like `scheduleConsumer`?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Avoid resource deadlocks for finite stream jobs when resources are limited
> --
>
> Key: FLINK-10945
> URL: https://issues.apache.org/jira/browse/FLINK-10945
> Project: Flink
> Issue Type: Improvement
> Components: Distributed Coordination
>Affects Versions: 1.7.1
>Reporter: Zhu Zhu
>Assignee: Zhu Zhu
>Priority: Major
> Labels: pull-request-available
>
> Currently *resource deadlocks* can happen to finite stream jobs(or batch
> jobs) when resources are limited. In 2 cases as below:
> # Task Y is a pipelined downstream task of task X. When X takes all
> resources(slots), Y cannot acquire slots
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[
https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16711599#comment-16711599
]
ASF GitHub Bot commented on FLINK-10945:
zhuzhurk closed pull request #7252: [FLINK-10945] Add an
InputDependencyConstraint to avoid resource dead…
URL: https://github.com/apache/flink/pull/7252
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
index 3b3132b3d11..6def1dfa488 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
@@ -159,6 +159,9 @@
/** Determines if a task fails or not if there is an error in writing
its checkpoint data. Default: true */
private boolean failTaskOnCheckpointError = true;
+ /** The input dependency constraint to schedule tasks. */
+ private InputDependencyConstraint inputDependencyConstraint =
InputDependencyConstraint.ANY;
+
// --- User code values
private GlobalJobParameters globalJobParameters;
@@ -518,6 +521,30 @@ public ExecutionMode getExecutionMode() {
return executionMode;
}
+ /**
+* Sets the input dependency constraint for vertex scheduling. It
indicates when a task
+* should be scheduled considering its inputs status.
+*
+* The default constraint is {@link InputDependencyConstraint#ANY}.
+*
+* @param inputDependencyConstraint The input dependency constraint.
+*/
+ public void setInputDependencyConstraint(InputDependencyConstraint
inputDependencyConstraint) {
+ this.inputDependencyConstraint = inputDependencyConstraint;
+ }
+
+ /**
+* Gets the input dependency constraint for vertex scheduling. It
indicates when a task
+* should be scheduled considering its inputs status.
+*
+* The default constraint is {@link InputDependencyConstraint#ANY}.
+*
+* @return The input dependency constraint of this job.
+*/
+ public InputDependencyConstraint getInputDependencyConstraint() {
+ return inputDependencyConstraint;
+ }
+
/**
* Force TypeExtractor to use Kryo serializer for POJOS even though we
could analyze as POJO.
* In some cases this might be preferable. For example, when using
interfaces
diff --git
a/flink-core/src/main/java/org/apache/flink/api/common/InputDependencyConstraint.java
b/flink-core/src/main/java/org/apache/flink/api/common/InputDependencyConstraint.java
new file mode 100644
index 000..c7a1e6a3519
--- /dev/null
+++
b/flink-core/src/main/java/org/apache/flink/api/common/InputDependencyConstraint.java
@@ -0,0 +1,35 @@
+/*
+ * 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.api.common;
+
+/**
+ * This constraint indicates when a task should be scheduled considering its
inputs status.
+ */
+public enum InputDependencyConstraint {
+
+ /**
+* Schedule the task if any input is consumable.
+*/
+ ANY,
+
+ /**
+* Schedule the task if all the inputs are consumable.
+*/
+ ALL
+}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptor.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptor.java
index 7c2b30db32a..15bdb36f1d3 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptor.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptor.java
@@ -101,7 +101,8 @@ public String
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16711598#comment-16711598 ] ASF GitHub Bot commented on FLINK-10945: zhuzhurk commented on issue #7252: [FLINK-10945] Add an InputDependencyConstraint to avoid resource dead… URL: https://github.com/apache/flink/pull/7252#issuecomment-444913214 > Please open the PR against the master branch. Thanks Zentol for pointing out the fault. The new PR is https://github.com/apache/flink/pull/7255 This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > Labels: pull-request-available > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > > The design doc is here. > [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16711595#comment-16711595 ] ASF GitHub Bot commented on FLINK-10945: zhuzhurk opened a new pull request #7255: [FLINK-10945] Use InputDependencyConstraint to avoid resource dead… URL: https://github.com/apache/flink/pull/7255 …locks in LAZY_FROM_SOURCES scheduling when resources are limited ## What is the purpose of the change This PR add a job config InputDependencyConstraint, which helps to avoid resource deadlocks in LAZY_FROM_SOURCES scheduling when resources are limited, as described in [FLINK-10945](https://issues.apache.org/jira/browse/FLINK-10945). ## Brief change log - *Add InputDependencyConstraint to ExecutionConfig* - *Adjust isConsumable interface in IntermediateResultPartition to fit for the data actual consumable definition* - Change current execution lazy scheduling logic(in Execution.scheduleOrUpdateConsumers(edges)) to schedule tasks only if the InputDependencyConstraint is satisfied(an interface ExecutionVertex.checkInputDependencyConstraints is added to serve this purpose). ## Verifying this change This change added tests and can be verified as follows: - *Added IntermediateResultPartitionTest to validate IntermediateResultPartition changes* - *Added ExecutionVertexInputConstraintTest to validate the constraint check logic in ExecutionVertex* ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes) - The serializers: (no) - The runtime per-record code paths (performance sensitive): (no) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes) - The S3 file system connector: (no) ## Documentation - Does this pull request introduce a new feature? (yes) - If yes, how is the feature documented? (JavaDocs) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > Labels: pull-request-available > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > > The design doc is here. > [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16711471#comment-16711471 ] ASF GitHub Bot commented on FLINK-10945: zentol commented on issue #7252: [FLINK-10945] Add an InputDependencyConstraint to avoid resource dead… URL: https://github.com/apache/flink/pull/7252#issuecomment-444876798 Please open the PR against the master branch. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > Labels: pull-request-available > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > > The design doc is here. > [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16711363#comment-16711363 ] ASF GitHub Bot commented on FLINK-10945: zhuzhurk opened a new pull request #7252: [FLINK-10945] Add an InputDependencyConstraint to avoid resource dead… URL: https://github.com/apache/flink/pull/7252 …locks for finite stream jobs when resources are limited ## What is the purpose of the change This PR add a job config InputDependencyConstraint, which helps to avoid resource deadlocks in scheduling when resources are limited, as described in [FLINK-10945](https://issues.apache.org/jira/browse/FLINK-10945). ## Brief change log - *Add InputDependencyConstraint to ExecutionConfig* - *Adjust isConsumable interface in IntermediateResultPartition to fit for the data actual consumable definition* - Change current execution lazy scheduling logic(in Execution.scheduleOrUpdateConsumers(edges)) to schedule tasks only if the InputDependencyConstraint is satisfied(an interface ExecutionVertex.checkInputDependencyConstraints is added to serve this purpose). ## Verifying this change This change added tests and can be verified as follows: - *Added IntermediateResultPartitionTest to validate IntermediateResultPartition changes* - *Added ExecutionVertexInputConstraintTest to validate the constraint check logic in ExecutionVertex* ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes) - The serializers: (no) - The runtime per-record code paths (performance sensitive): (no) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes) - The S3 file system connector: (no) ## Documentation - Does this pull request introduce a new feature? (yes) - If yes, how is the feature documented? (JavaDocs) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > Labels: pull-request-available > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > > The design doc is here. > [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16711338#comment-16711338 ] ASF GitHub Bot commented on FLINK-10945: zhuzhurk opened a new pull request #7250: [FLINK-10945] Add an InputDependencyConstraint to avoid resource dead… URL: https://github.com/apache/flink/pull/7250 …locks for finite stream jobs when resources are limited ## What is the purpose of the change *(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)* ## Brief change log *(for example:)* - *The TaskInfo is stored in the blob store on job creation time as a persistent artifact* - *Deployments RPC transmits only the blob storage reference* - *TaskManagers retrieve the TaskInfo from the blob cache* ## Verifying this change *(Please pick either of the following options)* This change is a trivial rework / code cleanup without any test coverage. *(or)* This change is already covered by existing tests, such as *(please describe tests)*. *(or)* This change added tests and can be verified as follows: *(example:)* - *Added integration tests for end-to-end deployment with large payloads (100MB)* - *Extended integration test for recovery after master (JobManager) failure* - *Added test that validates that TaskInfo is transferred only once across recoveries* - *Manually verified the change by running a 4 node cluser with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.* ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (yes / no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no) - The serializers: (yes / no / don't know) - The runtime per-record code paths (performance sensitive): (yes / no / don't know) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know) - The S3 file system connector: (yes / no / don't know) ## Documentation - Does this pull request introduce a new feature? (yes / no) - If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > Labels: pull-request-available > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > > The design doc is here. >
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[
https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16711340#comment-16711340
]
ASF GitHub Bot commented on FLINK-10945:
zhuzhurk closed pull request #7250: [FLINK-10945] Add an
InputDependencyConstraint to avoid resource dead…
URL: https://github.com/apache/flink/pull/7250
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
index 3b3132b3d11..6def1dfa488 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
@@ -159,6 +159,9 @@
/** Determines if a task fails or not if there is an error in writing
its checkpoint data. Default: true */
private boolean failTaskOnCheckpointError = true;
+ /** The input dependency constraint to schedule tasks. */
+ private InputDependencyConstraint inputDependencyConstraint =
InputDependencyConstraint.ANY;
+
// --- User code values
private GlobalJobParameters globalJobParameters;
@@ -518,6 +521,30 @@ public ExecutionMode getExecutionMode() {
return executionMode;
}
+ /**
+* Sets the input dependency constraint for vertex scheduling. It
indicates when a task
+* should be scheduled considering its inputs status.
+*
+* The default constraint is {@link InputDependencyConstraint#ANY}.
+*
+* @param inputDependencyConstraint The input dependency constraint.
+*/
+ public void setInputDependencyConstraint(InputDependencyConstraint
inputDependencyConstraint) {
+ this.inputDependencyConstraint = inputDependencyConstraint;
+ }
+
+ /**
+* Gets the input dependency constraint for vertex scheduling. It
indicates when a task
+* should be scheduled considering its inputs status.
+*
+* The default constraint is {@link InputDependencyConstraint#ANY}.
+*
+* @return The input dependency constraint of this job.
+*/
+ public InputDependencyConstraint getInputDependencyConstraint() {
+ return inputDependencyConstraint;
+ }
+
/**
* Force TypeExtractor to use Kryo serializer for POJOS even though we
could analyze as POJO.
* In some cases this might be preferable. For example, when using
interfaces
diff --git
a/flink-core/src/main/java/org/apache/flink/api/common/InputDependencyConstraint.java
b/flink-core/src/main/java/org/apache/flink/api/common/InputDependencyConstraint.java
new file mode 100644
index 000..c7a1e6a3519
--- /dev/null
+++
b/flink-core/src/main/java/org/apache/flink/api/common/InputDependencyConstraint.java
@@ -0,0 +1,35 @@
+/*
+ * 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.api.common;
+
+/**
+ * This constraint indicates when a task should be scheduled considering its
inputs status.
+ */
+public enum InputDependencyConstraint {
+
+ /**
+* Schedule the task if any input is consumable.
+*/
+ ANY,
+
+ /**
+* Schedule the task if all the inputs are consumable.
+*/
+ ALL
+}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptor.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptor.java
index 7c2b30db32a..15bdb36f1d3 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptor.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputChannelDeploymentDescriptor.java
@@ -101,7 +101,8 @@ public String
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16707792#comment-16707792 ] Andrey Zagrebin commented on FLINK-10945: - Thanks, [~zhuzh], looks good! I started some discussions there. > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > > The design doc is here. > [https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16706811#comment-16706811 ] Zhu Zhu commented on FLINK-10945: - Here's the design doc and welcome comments. https://docs.google.com/document/d/1jpqC7OW_nLOSVOg06_QCWelicVtV6Au0Krg5m_S4kjY/edit?usp=sharing > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > Here one input corresponds to an *IntermediateResult*. The constraint can be > configured for *certain vertex or job-wide*. When the constraint is *ALL*, a > task will be launched after all its upstream tasks have been deployed, > avoiding case 2 of resource deadlocks. By setting all job edges to *BLOCKING* > and all vertex InputDependencyConstraint to *ALL*, we can make sure that all > tasks can finish once they are deployed, thus no resource deadlock will > happen. > Making the constraint *ANY* job-wide will keep the job scheduling behaves in > the same way as current version. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16704415#comment-16704415 ] Zhu Zhu commented on FLINK-10945: - Thanks [~azagrebin] for your attention. I will post the design later today or tomorrow. > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > Here one input corresponds to an *IntermediateResult*. The constraint can be > configured for *certain vertex or job-wide*. When the constraint is *ALL*, a > task will be launched after all its upstream tasks have been deployed, > avoiding case 2 of resource deadlocks. By setting all job edges to *BLOCKING* > and all vertex InputDependencyConstraint to *ALL*, we can make sure that all > tasks can finish once they are deployed, thus no resource deadlock will > happen. > Making the constraint *ANY* job-wide will keep the job scheduling behaves in > the same way as current version. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (FLINK-10945) Avoid resource deadlocks for finite stream jobs when resources are limited
[ https://issues.apache.org/jira/browse/FLINK-10945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16702103#comment-16702103 ] Andrey Zagrebin commented on FLINK-10945: - Hi [~zhuzh], the idea sounds really good. Do you plan to implement this improvement? If so, could you post the high level design for this change when it is available? > Avoid resource deadlocks for finite stream jobs when resources are limited > -- > > Key: FLINK-10945 > URL: https://issues.apache.org/jira/browse/FLINK-10945 > Project: Flink > Issue Type: Improvement > Components: Distributed Coordination >Affects Versions: 1.7.1 >Reporter: Zhu Zhu >Assignee: Zhu Zhu >Priority: Major > > Currently *resource deadlocks* can happen to finite stream jobs(or batch > jobs) when resources are limited. In 2 cases as below: > # Task Y is a pipelined downstream task of task X. When X takes all > resources(slots), Y cannot acquire slots to start, thus the back pressure > will block X to finish > # Task Y is a upstream task of task X. When X takes all resources(slots) and > Y cannot start, X cannot finish as some of its inputs are not finished > > We can avoid case 1 by setting all edges to be BLOCKING to avoid pipeline > back pressure. However, case 2 cannot be avoided as X(downstream task) will > be launched when any of its input result is ready. > To be detailed, say task X has BLOCKING upstream task Y and Z, X can be > launched when Z finishes, though task Y is not launched yet. This pre-launch > behaviour can be beneficial when there are plenty of resources, thus X can > process data from Z earlier before Y finishes its data processing. However, > resource deadlocks may happen when the resources are limited, e.g. in small > sessions. > > I’d propose introducing a constraint named as *InputDependencyConstraint* to > control the scheduling of vertices. It has 2 values: > # *ANY*. The vertex can be scheduled when any of its inputs is consumable. > # *ALL*. The vertex can be scheduled when all of its inputs are consumable. > Here one input corresponds to an *IntermediateResult*. The constraint can be > configured for *certain vertex or job-wide*. When the constraint is *ALL*, a > task will be launched after all its upstream tasks have been deployed, > avoiding case 2 of resource deadlocks. By setting all job edges to *BLOCKING* > and all vertex InputDependencyConstraint to *ALL*, we can make sure that all > tasks can finish once they are deployed, thus no resource deadlock will > happen. > Making the constraint *ANY* job-wide will keep the job scheduling behaves in > the same way as current version. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
