Ah, I see my disconnect now.
The local PR I had was not updated with commit 1d6da02 - hence the bad perf
numbers I observed (it takes a couple of days to finish) : and commented
based on analysis of the local code - my apologies !
Preliminary results seem to indicate reasonably close enough numbers -
thanks for clarifying.


On Sat, Aug 17, 2013 at 10:23 PM, 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
>
> Mridul, as I said, please try it with my change and
> spark.locality.wait.node = 0. If you want the exact same behavior as your
> old code, you can also set spark.locality.wait.process=0. The behavior will
> be *exactly* the same: - When multiple offers come in, Spark will look at
> *all* of them together and try to assign only process-local tasks first,
> then node-local, then rack-local, etc. Note that I added the logic to go
> through these levels in order across the set of offers. - If you set both
> node and process waits to 0, it will immediately jump up to rack to fill in
> all the remaining spots. I really don't see how it's different, except that
> I'm doing this with significantly simpler code.
> … <#1408d31ed1554ae1_>
> I've been working on similar stuff in Hadoop for a long time (wrote the
> Hadoop Fair Scheduler in 2008, which uses the same algorithm for locality
> and has been running fine on 1000+ node clusters since then at Facebook,
> many Cloudera users, etc), so it's not like this is an untested change.
> There was definitely a problem before for large clusters where you don't
> have executors on most of the data-containing nodes, but the combination of
> rack-awareness and the ability to configure these delays and to optimize
> across a set of offers fixes that. I'm not trying to remove code because I
> don't understand it -- I just think that it's important for this logic (and
> all of Spark's scheduling code) to be dead-simple in order to be easy to
> optimize and understandable to new developers.
>  Let me clarify the performance issue we face, in the current code, if we
> set node locality wait to 0 - the scaling is extremely bad due to heavy
> data movement at job start (it is worse than non zero node locality value
> actually - so it is pointless to set that atleast in our expts).
> Particularly when dataset is large. Effectively, almost all schedules end
> up being rack local. Setting it to anything non zero (however small - in
> our expts even 1ms - since effectively it is a much higher wait), the
> cluster utilization drops drastically. Before introducing this change, we
> had experimented with reducing locality wait to 0ms and 1ms in earlier
> codebase - and since that did not help matters, we modified the scheduling
> ... initially aggression was effectively any - which improved situation
> much better, and adding rack support improved it even further while
> minimizing io overhead but at slightly decreased cluster utilization (so it
> was a tradeoff betwe en both) . Scheduling is something we need to heavily
> experiment with - considering the non trivial gains we saw with relatively
> small changes.
>
> —
> Reply to this email directly or view it on GitHub<https://github.com/mesos/spark/pull/828/files#r5829281>
> .
>