flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Till Rohrmann (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-7153) Eager Scheduling can't allocate source for ExecutionGraph correctly
Date Mon, 16 Oct 2017 13:26:00 GMT

    [ https://issues.apache.org/jira/browse/FLINK-7153?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16205877#comment-16205877

Till Rohrmann commented on FLINK-7153:

Why don't we make the scheduling dependent on the future locations of the input tasks? This
would mean that the {{ExecutionVertex#getPreferredLocations}} would return a {{Collection<CompletableFuture<TaskManagerLocation>>}}
which contains the future locations of the inputs. When trying to allocate a slot for a given
task, the {{Scheduler}} will retrieve the future locations of the inputs, and only upon completion
it will allocate the slot for the requested task. This will ensure that we have all information
needed to calculate the scheduling. I've quickly prototyped a working version which you find
here [1].

The downside of this approach is that we lose the nice property that we only start deploying
tasks in eager mode if we are sure that we have enough slots available. The reason is that
we have to call {{Execution#deployToSlot}} in order to complete the location future of an
{{Execution}}. We can solve this problem, however, by separating the slot assignment and the
actual deploy call into two methods, e.g. {{Execution#assignSlot}} and {{Execution#deployToSlot}}.
I've prototyped this version here [2]. Before calling {{Execution#deployToSlot}} one has to
call explicitly {{Execution#tryAssignSlot}}. This complicates the API a little bit but it
will allow us to fail without deploying tasks in the eager deployment mode.

What do you think?

[1] https://github.com/tillrohrmann/flink/tree/fixGroupScheduling
[2] https://github.com/tillrohrmann/flink/tree/fixGroupScheduling2

> Eager Scheduling can't allocate source for ExecutionGraph correctly
> -------------------------------------------------------------------
>                 Key: FLINK-7153
>                 URL: https://issues.apache.org/jira/browse/FLINK-7153
>             Project: Flink
>          Issue Type: Bug
>          Components: JobManager
>    Affects Versions: 1.3.1
>            Reporter: Sihua Zhou
>            Assignee: Stephan Ewen
>            Priority: Blocker
>             Fix For: 1.4.0, 1.3.3
> The ExecutionGraph.scheduleEager() function allocate for ExecutionJobVertex one by one
via calling ExecutionJobVertex.allocateResourcesForAll(), here is two problem about it:
> 1. The ExecutionVertex.getPreferredLocationsBasedOnInputs will always return empty, cause
`sourceSlot` always be null until `ExectionVertex` has been deployed via 'Execution.deployToSlot()'.
So allocate resource base on prefered location can't work correctly, we need to set the slot
info for `Execution` as soon as Execution.allocateSlotForExecution() called successfully?
> 2. Current allocate strategy can't allocate the slot optimize.  Here is the test case:
> {code}
> JobVertex v1 = new JobVertex("v1", jid1);
> JobVertex v2 = new JobVertex("v2", jid2);
> SlotSharingGroup group = new SlotSharingGroup();
> v1.setSlotSharingGroup(group);
> v2.setSlotSharingGroup(group);
> v1.setParallelism(2);
> v2.setParallelism(4);
> v1.setInvokableClass(BatchTask.class);
> v2.setInvokableClass(BatchTask.class);
> v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED_BOUNDED);
> {code}
> Currently, after allocate for v1,v2, we got a local partition and three remote partition.
But actually, it should be 2 local partition and 2 remote partition. 
> The causes of the above problems is becuase that the current allocate strategy is allocate
the resource for execution one by one(if the execution can allocate from SlotGroup than get
it, Otherwise ask for a new one for it). 
> If we change the allocate strategy to two step will solve this problem, below is the
Pseudo code:
> {code}
> for (ExecutionJobVertex ejv: getVerticesTopologically) {
> //step 1: try to allocate from SlothGroup base on inputs one by one (which only allocate
resource base on location).
> //step 2: allocate for the remain execution.
> }
> {code}

This message was sent by Atlassian JIRA

View raw message