qlong commented on code in PR #53914:
URL: https://github.com/apache/spark/pull/53914#discussion_r2737235219
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -625,18 +674,48 @@ private[spark] class Executor(
(accums, accUpdates)
}
+ /**
+ * Obtains an IsolatedSessionState for the given job artifact state.
+ * Gets or creates a session from the cache, then acquires it. We need to
retry the cache
+ * lookup if the session was evicted between get() and acquire(). This can
happen when the
+ * cache is full and another task triggers eviction.
+ */
+ private def obtainSession(jobArtifactState: JobArtifactState):
IsolatedSessionState = {
+ var session: IsolatedSessionState = null
+ var acquired = false
+ while (!acquired) {
+ // Get or create session. The loader uses sessions map as the
authoritative store.
+ // This ensures there's only one IsolatedSessionState per UUID at any
time.
+ session = isolatedSessionCache.get(jobArtifactState.uuid, () => {
+ // Check the authoritative sessions map first. tryUnEvict() will
block if
Review Comment:
I agree it is ok to wait for the small IO to finish, it would be much more
complicated to make cleanup() noblocking.
--
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]
