sunchao commented on code in PR #56516:
URL: https://github.com/apache/spark/pull/56516#discussion_r3607136164
##########
core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala:
##########
@@ -160,19 +190,4 @@ private[spark] class ResourceProfileManager(sparkConf:
SparkConf,
readLock.unlock()
}
}
-
- /*
- * If the ResourceProfile passed in is equivalent to an existing one, return
the
- * existing one, other return None
- */
- def getEquivalentProfile(rp: ResourceProfile): Option[ResourceProfile] = {
Review Comment:
[P2] Keep explicit profiles distinct from the actual default profile
`resourcesEqual` compares the resource maps and runtime class, but that is
not a complete equivalence relation for ID 0. Kubernetes applies
`spark.kubernetes.executor.request.cores`, CPU limits, and pod-template
resource handling only to the default profile; YARN also propagates
default-only `spark.yarn.executor.resource.*` settings.
For example, with `spark.executor.cores=4` and
`spark.kubernetes.executor.request.cores=0.5`, an explicit non-default profile
containing the same maps as the default requested 4 CPUs before this change. It
now resolves to the default object/ID 0 and requests 0.5 CPU, while also
acquiring the other default-only behavior. Please preserve whether the object
is the manager's actual default profile (object identity is safer than merely
testing ID 0), or compare complete effective cluster-manager requirements, and
add Kubernetes/YARN coverage for this case.
##########
core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala:
##########
@@ -139,13 +139,43 @@ private[spark] class ResourceProfileManager(sparkConf:
SparkConf,
}
// do this outside the write lock only when we add a new profile
if (putNewProfile) {
- // force the computation of maxTasks and limitingResource now so we
don't have cost later
- rp.limitingResource(sparkConf)
- logInfo(log"Added ResourceProfile id: ${MDC(LogKeys.RESOURCE_PROFILE_ID,
rp.id)}")
- listenerBus.post(SparkListenerResourceProfileAdded(rp))
+ onProfileAdded(rp)
}
}
+ /**
+ * Get the registered ResourceProfile whose resources are equal to the given
one, registering
+ * the given profile first if no equivalent one exists yet.
+ */
+ def getOrAddEquivalentProfile(rp: ResourceProfile): ResourceProfile = {
+ isSupported(rp)
+ var addedProfile: Option[ResourceProfile] = None
+ val resolvedProfile = {
+ writeLock.lock()
+ try {
+ resourceProfileIdToResourceProfile.collectFirst {
Review Comment:
[P1] This same-ID check is also required for correctness, not only for the
common-case performance path. `ResourceProfile.nextProfileId` starts at 0, so a
public profile constructed before `SparkContext` can receive ID 0; context
initialization then creates the real default profile and also forces its ID to
0. If the preconstructed profile is later passed to `RDD.withResources`, the
equivalence scan can miss and the unconditional `put(0, rp)` replaces the
manager's canonical default entry.
After that, `defaultResourceProfile` still points to the original object
while `resourceProfileFromId(0)` returns the custom one. Unrelated default
stages also use ID 0, so they inherit the custom CPU/GPU requirements; a GPU
request can make all ordinary stages unschedulable. The old
`addResourceProfile` path did not replace an occupied ID. Please preserve the
existing mapping (or reject the collision) and add an ID-0 regression.
##########
core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala:
##########
@@ -160,19 +190,4 @@ private[spark] class ResourceProfileManager(sparkConf:
SparkConf,
readLock.unlock()
}
}
-
- /*
- * If the ResourceProfile passed in is equivalent to an existing one, return
the
- * existing one, other return None
- */
- def getEquivalentProfile(rp: ResourceProfile): Option[ResourceProfile] = {
Review Comment:
[P2] Validate the profile before making it discoverable
The new entry is inserted at line 159 before `onProfileAdded` runs, but
`onProfileAdded` calls `limitingResource`, which can throw for publicly
constructible profiles—for example, executor GPU amount 1 with task GPU amount
2 (or executor cores 1 with task CPUs 2). The failed entry is never removed. A
separately constructed equivalent RDD profile, or an identical Spark Connect
retry, then matches the stored entry and returns successfully without
re-running validation or posting `SparkListenerResourceProfileAdded`.
Before this change, a distinct equivalent profile had a distinct ID and its
registration retried the validation, so it failed again. Please validate before
insertion, or otherwise ensure failed/partially initialized entries cannot be
observed, and add a two-attempt regression asserting that both equivalent
requests fail and no unannounced profile remains registered.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
