Skip to content

alternative #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Oct 2, 2019
Merged

alternative #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2fc0288
alternative
dongjoon-hyun Sep 25, 2019
04f931d
Recover comments
dongjoon-hyun Sep 25, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions ...n/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ import org.apache.spark.internal.config.Tests.IS_TESTING
import org.apache.spark.launcher.{LauncherBackend, SparkAppHandle}
import org.apache.spark.network.netty.SparkTransportConf
import org.apache.spark.network.shuffle.mesos.MesosExternalBlockStoreClient
import org.apache.spark.rpc.{RpcEndpointAddress, RpcEndpointRef}
import org.apache.spark.rpc.RpcEndpointAddress
import org.apache.spark.scheduler.{SlaveLost, TaskSchedulerImpl}
import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
import org.apache.spark.util.Utils
Expand Down Expand Up @@ -375,9 +375,18 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(

logDebug(s"Received ${offers.size} resource offers.")

// nodeBlacklist() currently only gets updated based on failures in spark tasks.
// If a mesos task fails to even start -- that is,
// if a spark executor fails to launch on a node -- nodeBlacklist does not get updated
// see SPARK-24567 for details
val blacklist = scheduler.nodeBlacklist()
val (matchedOffers, unmatchedOffers) = offers.asScala.partition { offer =>
val offerAttributes = toAttributeMap(offer.getAttributesList)
matchesAttributeRequirements(slaveOfferConstraints, offerAttributes)
if (blacklist.contains(offer.getHostname)) {
false
Copy link
Author

@dongjoon-hyun dongjoon-hyun Sep 25, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will save many computation on the invalid offers.

Copy link
Owner

@IgorBerman IgorBerman Sep 26, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't mind to add this optimisation, but imo, you can't remove original check. Since they happening in two different places and the code that matches offer is "repeated" at canLaunchTask

Copy link
Author

@dongjoon-hyun dongjoon-hyun Sep 26, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It sounds like you are saying unmatchedOffers will be evaluated at canLaunchTask. Does it what you mean really? At line 392, we simple do declineUnmatchedOffers(d, unmatchedOffers).

Copy link
Owner

@IgorBerman IgorBerman Sep 26, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mean that even though the offer is matched, then there is code in canLaunchTask that verifies that task can be launched with respect to cpu/memory requirements(so also blacklisting)

Copy link
Author

@dongjoon-hyun dongjoon-hyun Sep 26, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I said the computation on the invalid offers (not matched).

This will save many computation on the invalid offers.

Copy link

@squito squito Oct 1, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree with @dongjoon-hyun , seems this would be more efficient. I don't understand the objection

} else {
val offerAttributes = toAttributeMap(offer.getAttributesList)
matchesAttributeRequirements(slaveOfferConstraints, offerAttributes)
}
}

declineUnmatchedOffers(d, unmatchedOffers)
Expand Down Expand Up @@ -580,11 +589,6 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
cpus + totalCoresAcquired <= maxCores &&
mem <= offerMem &&
numExecutors < executorLimit &&
// nodeBlacklist() currently only gets updated based on failures in spark tasks.
// If a mesos task fails to even start -- that is,
// if a spark executor fails to launch on a node -- nodeBlacklist does not get updated
// see SPARK-24567 for details
!scheduler.nodeBlacklist().contains(offerHostname) &&
meetsPortRequirements &&
satisfiesLocality(offerHostname)
}
Expand Down
4 changes: 2 additions & 2 deletions ...la/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackendSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ import org.apache.spark.internal.config._
import org.apache.spark.network.shuffle.mesos.MesosExternalBlockStoreClient
import org.apache.spark.rpc.{RpcAddress, RpcEndpointRef}
import org.apache.spark.scheduler.TaskSchedulerImpl
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{RegisterExecutor}
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.RegisterExecutor
import org.apache.spark.scheduler.cluster.mesos.Utils._

class MesosCoarseGrainedSchedulerBackendSuite extends SparkFunSuite
Expand Down Expand Up @@ -126,7 +126,7 @@ class MesosCoarseGrainedSchedulerBackendSuite extends SparkFunSuite
// Offer resources from the same slave
offerResources(List(offer2))
// but since it's blacklisted the offer is declined
verifyDeclinedOffer(driver, createOfferId("o1"))
verifyDeclinedOffer(driver, createOfferId("o1"), true)
}

test("mesos supports spark.executor.cores") {
Expand Down