tgravescs commented on a change in pull request #25047:
[WIP][SPARK-27371][CORE] Support GPU-aware resources scheduling in Standalone
URL: https://github.com/apache/spark/pull/25047#discussion_r301136666
##########
File path:
core/src/main/scala/org/apache/spark/scheduler/ExecutorResourceInfo.scala
##########
@@ -17,85 +17,14 @@
package org.apache.spark.scheduler
-import scala.collection.mutable
-
-import org.apache.spark.SparkException
-import org.apache.spark.util.collection.OpenHashMap
+import org.apache.spark.resource.ResourceAllocator
/**
* Class to hold information about a type of Resource on an Executor. This
information is managed
* by SchedulerBackend, and TaskScheduler shall schedule tasks on idle
Executors based on the
* information.
- * Please note that this class is intended to be used in a single thread.
* @param name Resource name
* @param addresses Resource addresses provided by the executor
*/
-private[spark] class ExecutorResourceInfo(
- val name: String,
- addresses: Seq[String]) extends Serializable {
-
- /**
- * Map from an address to its availability, the value `true` means the
address is available,
- * while value `false` means the address is assigned.
- * TODO Use [[OpenHashMap]] instead to gain better performance.
- */
- private val addressAvailabilityMap = mutable.HashMap(addresses.map(_ ->
true): _*)
-
- /**
- * Sequence of currently available resource addresses.
- */
- def availableAddrs: Seq[String] = addressAvailabilityMap.flatMap { case
(addr, available) =>
- if (available) Some(addr) else None
- }.toSeq
-
- /**
- * Sequence of currently assigned resource addresses.
- * Exposed for testing only.
- */
- private[scheduler] def assignedAddrs: Seq[String] = addressAvailabilityMap
- .flatMap { case (addr, available) =>
- if (!available) Some(addr) else None
- }.toSeq
-
- /**
- * Acquire a sequence of resource addresses (to a launched task), these
addresses must be
- * available. When the task finishes, it will return the acquired resource
addresses.
- * Throw an Exception if an address is not available or doesn't exist.
- */
- def acquire(addrs: Seq[String]): Unit = {
- addrs.foreach { address =>
- if (!addressAvailabilityMap.contains(address)) {
- throw new SparkException(s"Try to acquire an address that doesn't
exist. $name address " +
- s"$address doesn't exist.")
- }
- val isAvailable = addressAvailabilityMap(address)
- if (isAvailable) {
- addressAvailabilityMap(address) = false
- } else {
- throw new SparkException(s"Try to acquire an address that is not
available. $name " +
- s"address $address is not available.")
- }
- }
- }
-
- /**
- * Release a sequence of resource addresses, these addresses must have been
assigned. Resource
- * addresses are released when a task has finished.
- * Throw an Exception if an address is not assigned or doesn't exist.
- */
- def release(addrs: Seq[String]): Unit = {
- addrs.foreach { address =>
- if (!addressAvailabilityMap.contains(address)) {
- throw new SparkException(s"Try to release an address that doesn't
exist. $name address " +
- s"$address doesn't exist.")
- }
- val isAvailable = addressAvailabilityMap(address)
- if (!isAvailable) {
- addressAvailabilityMap(address) = true
- } else {
- throw new SparkException(s"Try to release an address that is not
assigned. $name " +
- s"address $address is not assigned.")
- }
- }
- }
-}
+private[spark] class ExecutorResourceInfo(name: String, addresses: Seq[String])
Review comment:
this is going to conflict with what I did here:
https://github.com/apache/spark/pull/25037
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
