SPARK-1937: fix issue with task locality

[lirui-apache]

Don't check executor/host availability when creating a TaskSetManager. Because the executors may haven't been registered when the TaskSetManager is created, in which case all tasks will be considered "has no preferred locations", and thus losing data locality in later scheduling.

[AmplabJenkins]

Can one of the admins verify this patch?

[mridulm]

This is better handled in user code and not in scheduler.
We need a way to surface how many executors have registered - but once that is available, spinning until some minimum threshold of executors is available should be done in user code and then start computations.

Ofcourse, if there is interest, this could be abstracted out for common use in contrib.

Currently, for our jobs, we have an arbitrary sleep introduced which is based on expectation of cluster load.

One unstated impact of this, which is not mentioned in the PR description, is that default parallelism is determined (in yarn mode atleast) by the number of available executors.
So the number of reducers used can be very low resulting for the first (few) jobs - which might have changed 'later' when reducers actually kick in.

[rxin]

Tagging @kayousterhout ...

[kayousterhout]

Yeah I agree with @mridulm. It seems like, with this change, you're counting on the executors to finish registering before the delay threshold runs out or before all of the tasks have been run. It seems cleaner to just wait until enough executors have started before launching jobs.

[lirui-apache]

If I understand, the application cannot control how many executors to launch (at least with the standalone mode). So I suppose it's difficult for the user to determine when he/she gets "enough" executors.

Another possible way to solve this issue is that when the TaskScheduler finds new executors/hosts in the resource offer, it can inform the TaskSetManager to re-compute the pending lists. TaskSetManager currently handles event like "executor lost", maybe it should also handle "executor added"?

I have another concern about pendingTasksWithNoPrefs in TaskSetManager. This list now contains two kinds of tasks: tasks that truly have no preferred locations, and tasks whose preferred locations are not available. Tasks in this list are considered as PROCESS_LOCAL, which means they take precedence over tasks in NODE_LOCAL and RACK_LOCAL lists. Therefore in some scenario (the allowed locality is PROCESS_LOCAL), TaskSetManager may prefer tasks whose desired location is unavailable, to tasks that can run with some locality. Not sure if this behavior is intended?

[lirui-apache]

I've made some modifications, please help to see if this makes sense :)
@mridulm @rxin @kayousterhout

[kayousterhout]

Why do you filter out tasks that already have a copy running? Seems like you might as well use the better location in the event of task failure or speculation

[lirui-apache]

Actually this is the same filter used in "findTaskFromList". TaskSetManager will re-add failed task to pending lists, so it can have the updated executor availability. For speculative task, TaskSetManager won't check the availability of executors in "findSpeculativeTask". Please correct me if I misunderstand anything.

[kayousterhout]

It looks like when you re-add tasks after an executor is added, tasks that newly have one of their preferred locations available are not removed from the pendingTasksWithNoPrefs list? Doesn't that mean that the task's locality constraints can still get ignored?

[mateiz]

I'm curious, why are you adding a flag for this instead of always leaving it on? It seems like something we should have on all the time.

[mateiz]

Hei @lirui-intel, I talked about this a bit with Kay and we think there's a simpler and more general approach.

• In addPendingTask, always add the task to host, rack and executor-specific lists even if those executors aren't currently up (i.e. remove the current if statements that present that); then add it to pendingTasksWithNoPrefs only if there were no such executors up
• When a new executor comes up, remove any tasks from pendingTasksWithNoPrefs that have preferred locations on that executor / host / rack; now locality-aware scheduling will automatically happen for those

One other thing to watch out for will be the time complexity of these operations -- we don't want linear-time stuff in the size of pendingTasksWithNoPrefs if we can help it because there might be hundreds of executors coming up over time. But we can worry about that once we have an implementation.

This feature is great to have overall though because it will allow dynamic resizing of clusters (e.g. on YARN we might scale up and down based on whether we have pending task sets).

[lirui-apache]

Yes @mateiz great idea. One quick question is that tasks in pendingTasksWithNoPrefs are considered as PROCESS_LOCAL. Suppose we have no tasks in pendingTasksForExecutor, but some tasks in pendingTasksForHost which are NODE_LOCAL. And also some tasks in pendingTasksWithNoPrefs whose preferred locations are not available. Now if the current allowed locality is PROCESS_LOCAL, TaskSetManager will pick tasks from pendingTasksWithNoPrefs rather than pendingTasksForHost.

[mateiz]

I see, we might be able to fix that as a separate issue. We could count pendingTasksWithNoPrefs as being at a "special" level we try only after we try the real locality levels. But that seems independent of this patch, maybe you can report it as a separate JIRA.

[lirui-apache]

Hi @mateiz , I think we should distinguish between tasks that truly have no preference, and tasks whose preference is unavailable when picking tasks from pendingTasksWithNoPrefs. If a task really have no preference, it should indeed be considered as PROCESS_LOCAL because it runs equally well anywhere. And it should take precedence over NODE_LOCAL tasks. For tasks whose preference is unavailable, they should wait until we try all the real locality levels.
Maybe we should firstly figure out how to deal with this pendingTasksWithNoPrefs? Because otherwise it could be difficult to even verify if the patch works.

[kayousterhout]

My understanding is that the problem mentioned only exists when a job has a combination of tasks with constraints and tasks without constraints. AFAIK this is not a common case at all -- since a stage will either be the first in the job, in which case all tasks have locality constraints on the input data, or will be a reduce stage, in which case none of the tasks have locality constraints.

[lirui-apache]

That's true @kayousterhout , but my point is that what if all the tasks only specify NODE_LOCAL preference (common case when the RDD is created from some HDFS file)? Then tasks in pendingTasksWithNoPrefs (considered PROCESS_LOCAL) will get priority to run. I ran into this during the test for the patch.
Is it possible that we set some time out waiting for new executors to come up? Before the time out, only tasks without preference (if any) in pendingTasksWithNoPrefs can run as PROCESS_LOCAL. After this period, we believe it's unlikely new executors gonna come up, so all tasks in pendingTasksWithNoPrefs can be treated as PROCESS_LOCAL (pendingTasksWithNoPrefs can still be re-computed if new executors do come up later).

[lirui-apache]

I've revised the patch as @mateiz suggested: tasks will be added to corresponding lists even when preferred location is unavailable, in which case it'll also be added to pendingTasksWithNoPrefs. Later when new executor comes up, pendingTasksWithNoPrefs is re-computed to remove tasks that have available preferred locations.
I also added a "waiting for new executor" period. During that period, tasks in pendingTasksWithNoPrefs will wait until we tried all other locality levels (unless they really have no constraints). After this period, tasks left in pendingTasksWithNoPrefs will be scheduled as PROCESS_LOCAL.
Please help to review this, thanks.

[kayousterhout]

Is this something the application can just do: if it wants to wait 3 seconds before scheduling anything on non-local executors, just sleep for 3 seconds before trying to launch any jobs? I'm wary of adding more config options to the scheduler.

[lirui-apache]

This waiting period is only intended for pendingTasksWithNoPrefs. Suppose pendingTasksWithNoPrefs contains tasks whose preference is unavailable. Within this waiting period, we want to try pendingTasksForExecutor, pendingTasksForHost and pendingTasksForRack first because tasks in these lists do have some locality. And when an executor is added, we remove tasks newly have locality from pendingTasksWithNoPrefs. Then after the waiting period, we believe no executor will come for tasks still remain in pendingTasksWithNoPrefs. So they can be shceduled as PROCESS_LOCAL.
You can see tasks in pendingTasksForHost can still get scheduled even within the period. We're just holding back on pendingTasksWithNoPrefs. I think it's better than holding back the whole application and schedule nothing.

[mateiz]

It doesn't make sense to put this here because it will apply to every TaskSet, no matter how late into the application it was submitted, so you'll get a 3-second latency on every TaskSet that is missing one of its preferred nodes. Can we not add this as part of this patch, and simply make the change to put tasks in the node- and rack-local lists even if no nodes are available in those right now? Then later we can update the code that calls resourceOffer to treat tasks that have preferred locations but are missing executors for them specially.

[kayousterhout]

@lirui-intel I still don't understand the problem you mentioned: in TaskSetManager.findTask(), it tries to schedule process local, node local, and rack local tasks before looking at tasks from pendingTasksWithNoPrefs. So how can a task from pendingTasksWithNoPrefs be scheduled before a PROCESS_LOCAL one? (I see that tasks from pendingTasksWithNoPrefs are marked with locality level PROCESS_LOCAL when they're launched -- but this doesn't look like it affects the scheduling order)

[lirui-apache]

Hi @kayousterhout , let's consider a map stage whose tasks all have NODE_LOCAL preference. So pendingTasksForExecutor is empty and all tasks are added to pendingTasksForHost. If at the beginning no node is available, then all tasks will also be added to pendingTasksWithNoPrefs (as @mateiz suggested). When we look for a task to launch, we always try to launch a PROCESS_LOCAL one first. Suppose now an executor comes in and it can satisfy some tasks in pendingTasksForHost. However, since we always try with PROCESS_LOCAL first, pendingTasksForHost will simply be skipped in TaskSetManager.findTask. And we end up picking a task in pendingTasksWithNoPrefs. You can see there's an if statement when trying to pick tasks from pendingTasksForHost and pendingTasksForRack, to test if we currently allow such "low" locality level. But tasks in pendingTasksForExecutor and pendingTasksWithNoPrefs are picked unconditionally since they are considered have the highest level.

[mateiz]

Nit: should be indented with 2 more spaces:

if (!pendingTasksForExecutor.isEmpty && getLocalityWait(PROCESS_LOCAL) != 0 &&
    pendingTasksForExecutor.keySet.exists(sched.isExecutorAlive(_))) {
  levels += PROCESS_LOCAL
}

[mateiz]

Hey @lirui-intel, sorry for taking a while to get back to this, but it looks good to me. Iterating through only the TaskSet's valid locality levels sounds fine. I made some small comments on the code to improve it but once those are done it should be good to merge.

[lirui-apache]

Sorry about the code style and thanks @mateiz for pointing out. I've updated the patch.

[mateiz]

Jenkins, this is ok to test

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished. All automated tests passed.

[AmplabJenkins]

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16037/

[mateiz]

Thanks Rui. I've merged this in now.

[lirui-apache]

Thanks everybody :-)