dongjoon-hyun commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4339138611
##########
core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala:
##########
@@ -931,6 +934,351 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
completion
}
+ private def startDaemonThread(name: String)(body: => Unit): Thread = {
+ val thread = new Thread(() => body, name)
+ thread.setDaemon(true)
+ thread.start()
+ thread
+ }
+
+ private def joinThread(thread: Thread): Unit = {
+ thread.join(TimeUnit.MINUTES.toMillis(1))
+ assert(!thread.isAlive, s"${thread.getName} is still running")
+ }
+
+ /**
+ * Fails the task on a new thread, whose failure listener then holds the
listener invocation
+ * until the returned latch is counted down. Returns that latch and the
thread.
+ */
+ private def holdListenerInvocation(context: TaskContext): (CountDownLatch,
Thread) = {
+ val holding = new CountDownLatch(1)
+ val release = new CountDownLatch(1)
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable): Unit
= {
+ holding.countDown()
+ release.await(1, TimeUnit.MINUTES)
+ }
+ })
+ val holder = startDaemonThread("listener-holder") {
+ context.markTaskFailed(new Exception("failed on another thread"))
+ }
+ assert(holding.await(1, TimeUnit.MINUTES))
+ (release, holder)
+ }
+
+ /** Waits until `thread` waits in `markTaskCompleted()` for the listener
invocation. */
+ private def awaitHandoverWait(thread: Thread): Unit = {
+ eventually(timeout(1.minute)) {
+ assert(thread.getState == Thread.State.TIMED_WAITING)
+ }
+ }
+
+ test("SPARK-60222: a listener added as completion starts does not take over
the listeners") {
+ // A thread that sees the task completed and adds a listener at that
moment must not become
+ // the thread that runs the pending listeners: the completing thread
claims the listener
+ // invocation together with setting `completed`.
+ (1 to 500).foreach { _ =>
+ val context = TaskContext.empty()
+ val pendingListenerThread = new AtomicReference[Thread]()
+ context.addTaskCompletionListener[Unit] { _ =>
+ pendingListenerThread.set(Thread.currentThread())
+ }
+ val ready = new CountDownLatch(1)
+ val adder = startDaemonThread("adder") {
+ ready.countDown()
+ while (!context.isCompleted()) {}
+ context.addTaskCompletionListener[Unit] { _ => }
+ }
+ ready.await()
+ context.markTaskCompleted(None)
+ joinThread(adder)
+ assert(pendingListenerThread.get eq Thread.currentThread())
+ }
+ }
+
+ test("SPARK-60222: a failing listener does not let another thread take over
the listeners") {
+ // While a failing listener's failure listeners run, the thread invoking
the listeners must
+ // stay registered, so that another thread adding listeners at that moment
does not start
+ // running the pending ones, possibly concurrently with this thread.
+ (1 to 200).foreach { _ =>
+ val context = TaskContext.empty()
+ val pendingListenerThread = new AtomicReference[Thread]()
+ context.addTaskCompletionListener[Unit] { _ =>
+ pendingListenerThread.set(Thread.currentThread())
+ }
+ // Two failing listeners, which run first: the second failure finds the
task already failed.
+ (1 to 2).foreach { i =>
+ context.addTaskCompletionListener(new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit = throw
new Exception(s"$i")
+ })
+ }
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable):
Unit = {}
+ })
+ val ready = new CountDownLatch(1)
+ val adder = startDaemonThread("adder") {
+ ready.countDown()
+ while (!context.isCompleted()) {}
+ (1 to 200).foreach { _ => context.addTaskCompletionListener[Unit] { _
=> } }
+ }
+ ready.await()
+ intercept[TaskCompletionListenerException] {
+ context.markTaskCompleted(None)
+ }
+ joinThread(adder)
+ assert(pendingListenerThread.get eq Thread.currentThread())
+ }
+ }
+
+ test("SPARK-60222: completion waits for another thread invoking listeners
and runs them") {
+ val context = TaskContext.empty()
+ val listenerThreads = new ConcurrentLinkedQueue[(String, Thread)]()
+ context.addTaskCompletionListener[Unit] { _ =>
Review Comment:
**13. [Nit] Five new listeners rebuild `loggingListener`**
This listener and the ones at L1061, L1113, L1137 and L1217 are
`loggingListener(listenerThreads, "pending")` and so on (L900-907), which the
SPARK-60113 tests above already use. Could they use it? The "killer" setup at
L1043-1053 also repeats `holdListenerInvocation` except for the call that takes
the invocation, so a by-name parameter there could cover it too.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -280,18 +338,106 @@ private[spark] class TaskContextImpl(
exception
}
- private def invokeTaskCompletionListeners(error: Option[Throwable]): Unit = {
+ /**
+ * Waits until the thread invoking listeners hands the invocation over to
`currentThread`, which
+ * has registered itself as `nextListenerInvocationThread`.
+ *
+ * Interrupts do not stop the wait, and are passed on once it is done. The
wait of
+ * `PipelinedWriterTask.awaitExit()` in `BasePythonRunner` drops them
instead, because it is a
+ * listener itself and the listeners right after it must still be able to
clean up. This wait
+ * is not a listener but part of `markTaskCompleted()`, so it leaves the
interrupt status (e.g.
+ * set by a task kill) as the completion listeners would see it without the
wait.
+ *
+ * A failure to log the wait (e.g. an OutOfMemoryError) ends it, unlike the
wait of
+ * `PipelinedWriterTask.awaitExit()`, which must go on since the writer
would otherwise use
+ * memory that the next listeners free. Ending this wait early leaves the
completion listeners
+ * to the other thread, so they may run after the task has released its
memory, but it lets a
Review Comment:
**7. [Low] Ending the wait early does not always do what this says**
Two cases contradict "it lets a fatal error reach the executor" and "leaves
the completion listeners to the other thread":
- When another thread holds a failure invocation, the task body has usually
failed too, e.g. with `TaskKilledException` after a kill, or with the error
rethrown from `getTaskFailure`. Then `markTaskCompleted(Some(e))` runs inside
`runTaskWithListeners`'s `catch { case t: Throwable => e.addSuppressed(t) }`
(`TaskContext.scala:250-255`). An `OutOfMemoryError` from `logWarning` becomes
a suppressed exception of `e`, which `Executor.isFatalError` does not follow,
and the executor keeps running. The error reaches the executor only when the
task body succeeded.
- If the holder hands over after L376-377 and before a `logWarning` that
throws, it has skipped its leftover branch (L595) because this thread was
registered. The `finally` (L406-407) then releases the invocation to nobody,
and no thread runs the completion listeners.
Could the scaladoc limit the first claim to a successful task body, and
mention the second case?
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -280,18 +338,106 @@ private[spark] class TaskContextImpl(
exception
}
- private def invokeTaskCompletionListeners(error: Option[Throwable]): Unit = {
+ /**
+ * Waits until the thread invoking listeners hands the invocation over to
`currentThread`, which
+ * has registered itself as `nextListenerInvocationThread`.
+ *
+ * Interrupts do not stop the wait, and are passed on once it is done. The
wait of
+ * `PipelinedWriterTask.awaitExit()` in `BasePythonRunner` drops them
instead, because it is a
+ * listener itself and the listeners right after it must still be able to
clean up. This wait
+ * is not a listener but part of `markTaskCompleted()`, so it leaves the
interrupt status (e.g.
+ * set by a task kill) as the completion listeners would see it without the
wait.
+ *
+ * A failure to log the wait (e.g. an OutOfMemoryError) ends it, unlike the
wait of
+ * `PipelinedWriterTask.awaitExit()`, which must go on since the writer
would otherwise use
+ * memory that the next listeners free. Ending this wait early leaves the
completion listeners
+ * to the other thread, so they may run after the task has released its
memory, but it lets a
+ * fatal error reach the executor, which then exits instead of waiting for
that thread.
+ */
+ private def awaitListenerInvocationHandover(currentThread: Thread): Unit = {
+ val startNs = System.nanoTime()
+ var interrupted = false
+ var warned = false
+ var handedOver = false
+ try {
+ while (!handedOver) {
+ val holder = synchronized {
+ val deadlineNs = System.nanoTime() +
+ TimeUnit.MILLISECONDS.toNanos(math.max(1L,
listenerHandoverWarnIntervalMs))
+ var remainingNs = deadlineNs - System.nanoTime()
+ while (!listenerInvocationThread.contains(currentThread) &&
remainingNs > 0) {
+ try {
+ wait(math.max(1L, TimeUnit.NANOSECONDS.toMillis(remainingNs)))
+ } catch {
+ case _: InterruptedException => interrupted = true
+ }
+ remainingNs = deadlineNs - System.nanoTime()
+ }
+ handedOver = listenerInvocationThread.contains(currentThread)
+ if (handedOver) None else listenerInvocationThread
+ }
+ // Outside the lock: walking the stack of another thread stops it at a
safepoint, and
+ // logging may block, which would otherwise hold up every thread using
this TaskContext.
+ // The holder does not change during the wait, since it hands the
invocation to this
+ // thread, so its stack trace is attached to the first warning only.
+ holder.foreach { holder =>
+ val waitedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNs)
+ val message = log"Task ${MDC(TASK_ATTEMPT_ID, taskAttemptId)} has
been waiting for " +
+ log"${MDC(TOTAL_TIME, waitedMs)} ms to run its completion
listeners until thread " +
+ log"${MDC(THREAD_NAME, holder.getName)} finishes running its task
listeners."
+ if (warned) {
+ logWarning(message)
+ } else {
+ val holderStack = new Exception(s"Stack trace of thread
${holder.getName}")
+ holderStack.setStackTrace(holder.getStackTrace)
Review Comment:
**9. [Nit] The holder's stack is not re-checked after the lock is released**
If the holder hands over between L377 and here and goes back to its own work
(the RPC dispatcher after `Task.kill`, the task reaper, or a pool thread), this
attaches those unrelated frames to the only warning that carries a stack, and
the wait then ends. `PipelinedWriterTask` re-checks after taking the stack
(`PythonRunner.scala:427-428`, `if (exited.getCount > 0)`).
`listenerInvocationThread` is volatile and can now only move from the holder to
this thread, so could this check `listenerInvocationThread.contains(holder)`
after `getStackTrace` and skip the warning otherwise?
--
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]
