LuciferYang commented on code in PR #13166:
URL: https://github.com/apache/gluten/pull/13166#discussion_r4266249593
##########
gluten-core/src/main/scala/org/apache/spark/task/TaskResources.scala:
##########
@@ -339,6 +358,7 @@ class TaskResourceRegistry extends Logging {
}
priorityToResourcesMapping.clear()
resources.clear()
+ released = true
Review Comment:
Fixed in a1fa79219: `releaseAll()` now sets `released` in a `finally`, so
the registry is marked released even when a release callback throws a fatal
error. The new test "registering after a fatal release failure fails" throws a
`LinkageError` from a release callback and checks that the late registration is
rejected.
##########
gluten-core/src/test/scala/org/apache/gluten/task/TaskResourceSuite.scala:
##########
@@ -119,4 +122,95 @@ class TaskResourceSuite extends AnyFunSuite with SQLHelper
{
// The registry was cleared on the way out, so a fresh task runs cleanly.
TaskResources.runUnsafe(assert(TaskResources.inSparkTask()))
}
+
+ test("Run unsafe - release callbacks run outside the global registry lock") {
+ // On a real executor the release callbacks include JNI native teardown
that
+ // can take milliseconds. Running them under the JVM-global registry lock
+ // serialized every concurrent task's completion and registration on the
+ // slowest callback; a concurrent task start must not block on it.
+ var concurrentTaskStarted = false
+ // Captured on the starter thread; read on this thread only after the join
+ // below confirms the thread terminated (the !isAlive assert gates it),
which
+ // establishes the happens-before to read these plain locals safely.
+ var starterFailure: Option[Throwable] = None
+ TaskResources.runUnsafe {
+ TaskResources.addResource(
+ UUID.randomUUID().toString,
+ new TaskResource {
+ override def release(): Unit = {
+ val starter = new Thread(
+ () => {
+ try {
+ TaskResources.runUnsafe {
+ TaskResources.addResource(
+ UUID.randomUUID().toString,
+ new TaskResource {
+ override def release(): Unit = {}
+ override def resourceName(): String = "concurrent task
resource"
+ }
+ )
+ }
+ concurrentTaskStarted = true
+ } catch {
+ case NonFatal(t) => starterFailure = Some(t)
+ }
+ })
+ starter.start()
+ starter.join(30000)
+ assert(
+ !starter.isAlive,
+ "concurrent task start is still blocked after the release
callback returned")
+ }
+ override def resourceName(): String = "slow releaser"
+ }
+ )
+ }
+ // Surface a genuine unrelated failure with its real stack instead of the
+ // misleading "blocked" message below.
+ starterFailure.foreach(t => throw t)
+ assert(concurrentTaskStarted, "concurrent task start was blocked by the
release pass")
+ }
+
+ test("Run unsafe - registering into a released registry fails") {
+ // A thread sharing the task's TaskContext can fetch the registry while
the release
+ // pass runs outside the global lock. Its registration must fail loudly
instead of
+ // landing in the cleared registry, where nothing would ever release it.
+ var lateFailure: Option[Throwable] = None
+ var lateReleased = false
+ var child: Thread = null
+ TaskResources.runUnsafe {
+ val tc = TaskContext.get()
+ TaskResources.addResource(
+ UUID.randomUUID().toString,
+ new TaskResource {
+ override def release(): Unit = {
+ child = new Thread(
+ () => {
+ SparkTaskUtil.setTaskContext(tc)
+ try {
+ TaskResources.addResource(
+ UUID.randomUUID().toString,
+ new TaskResource {
+ override def release(): Unit = lateReleased = true
+ override def resourceName(): String = "late resource"
+ })
+ } catch {
+ case NonFatal(t) => lateFailure = Some(t)
+ } finally {
+ SparkTaskUtil.unsetTaskContext()
+ }
+ })
+ child.start()
+ // Simulate a slow native teardown.
+ Thread.sleep(500)
Review Comment:
Addressed in a1fa79219. The test no longer relies on a fixed sleep. It waits
until the registering thread is blocked on the registry's monitor, and it
checks for the "already released" message, so the `IllegalStateException` from
a missing map entry can't satisfy it.
--
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]
