Hmm, on the contrary, the aggression logic is supposed to make it user
controllable on whether the tradeoff of waiting for optimal scheduling vs
maximising cluster utilization while maximising locality on the set of one
or more offers.
Particularly with locality wait and larger number of workers, the
reviveOffers thread will be invoking the method with nontrivial number of
offers - optimizing the single offer case was left for later (see below).

In our tests (ml apps with disk or mem ser on non trivial input datasets)
introduction of this one change speedup of jobs by 2.5x - 4x (iirc) with no
other change in system : with setting it to the default having no perf
change w.r.t earlier perf of jobs in our measurements.
As I mentioned before, idling large number of workers is extremely
suboptimal for our usecases.
I am sure for a large number of other applications of spark this is not a
concern : if entire working set fits in memory and is hosted in non-ser
mode, etc - which is why it would be good to allow its customization via
flags - while keeping it at defaults to handle common case.

Re single worker case : we can simply disable aggression code for this
(earlier, the last iteration expanded the locality constraint) when in this
last iteration and num offers = 1 : this was not done since reviveOffers
thread was optional unlike now.
This should disallow non host or instance local scheduling until timeout
for single offer case.

Regards
Mridul
 On Aug 16, 2013 3:33 AM, "Matei Zaharia" <notifications@github.com> wrote:

> In core/src/main/scala/spark/scheduler/cluster/ClusterScheduler.scala:
>
> >          for (i <- 0 until offers.size) {
> > -          val hostPort = offers(i).hostPort
> > -          // DEBUG code
> > -          Utils.checkHostPort(hostPort)
> > -
> > -          val numProcessLocalTasks =  math.max(0, math.min(manager.numPendingTasksForHostPort(hostPort), availableCpus(i)))
> > -          if (numProcessLocalTasks > 0){
> > -            val list = processLocalOffers.getOrElseUpdate(hostPort, new ArrayBuffer[Int])
> > -            for (j <- 0 until numProcessLocalTasks) list += i
> > +          val execId = offers(i).executorId
> > +          val host = offers(i).host
> > +          for (task <- manager.resourceOffer(execId, host, availableCpus(i))) {
> > +            tasks(i) += task
>
> Yes, I'll implement point a), but for point b), as I said above, it's not
> a good idea. Single worker offers are the common case while a job is
> running (e.g. as it finishes the first series of tasks and launches
> another), and it fundamentally wouldn't make sense to delay them to wait
> for multiple workers to finish because then you'd be wasting cluster
> resources. It's not just an issue when a job begins -- this aggression
> logic degrades locality throughout the job runtime. For example, if you
> have a job with cached data (process-local) that is running multiple tasks
> per core, some tasks will be scheduled non-locally in the current code,
> leading to a very expensive transfer of a block across nodes when you
> could've just waited a few of seconds to run on the right node. With my new
> code, it's still very easy to minimize the initial wait if that's a
> concern, and you get the additional bonus of making the process-local wait
> time different from the node/rack one.
>
> —
> Reply to this email directly or view it on GitHub<https://github.com/mesos/spark/pull/828/files#r5802888>
> .
>