dongjoon-hyun commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4371913161
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -412,21 +596,62 @@ private[spark] class TaskContextImpl(
// [[markTaskFailed]] here will have no effect and we simply
resume running the
// remaining completion handlers.
try {
- listenerInvocationThread = None
- markTaskFailed(e)
+ markTaskFailed(e, Nested)
} catch {
case t: Throwable => e.addSuppressed(t)
- } finally {
- synchronized {
- if (listenerInvocationThread.isEmpty) {
- listenerInvocationThread = Some(Thread.currentThread())
- }
- }
}
listenerExceptions += e
logError(log"Error in ${MDC(LISTENER, name)}", e)
}
}
+
+ def runLeftListeners(invoke: () => Unit): Unit = {
+ try {
+ invoke()
+ } catch {
+ // Merge the listeners' own failures, which the nested invocation has
logged, not the
+ // exception that wraps them. Any other error comes from that
invocation itself (e.g. an
+ // OutOfMemoryError) and propagates, through the release below.
+ case e: TaskCompletionListenerException => listenerExceptions ++=
e.getSuppressed.toSeq
+ }
+ }
+
+ // Chooses the next thing to do under the TaskContext lock, to be done
outside of it.
+ def nextStep(): Option[() => Unit] = synchronized {
+ if (!invokesFailureListeners && failureCauseOpt.isDefined &&
!onFailureCallbacks.empty()) {
+ val cause = failureCauseOpt.get
+ Some(() => runLeftListeners(() => invokeTaskFailureListeners(cause,
Nested)))
+ } else if (!listeners.empty()) {
+ val listener = listeners.pop()
+ Some(() => runListener(listener))
+ } else if (mode != Nested && nextListenerInvocationThread.isEmpty &&
completed &&
Review Comment:
**1. [Low] Completion listeners left to another thread can release a
broadcast's block lock under the wrong task**
The answer to [round-1 item
5](https://github.com/apache/spark/pull/59663#discussion_r4339138517) documents
that listeners left to this thread "must use the context passed to them"
(L513-516). An in-tree listener cannot do that when the code it calls reads the
thread-local. `TorrentBroadcast.releaseBlockManagerLock` registers `_ =>
blockManager.releaseLock(blockId)` (`TorrentBroadcast.scala:323`),
`BlockManager.releaseLock` passes no task to `BlockInfoManager.unlock`
(`BlockManager.scala:1550-1558`), and `unlock` then takes the task from
`TaskContext.get()`. That is `NON_TASK_WRITER` on a kill thread, and on the
task thread once `Task.run` has unset its context
(`BlockInfoManager.scala:286-287`).
So this branch can run the release listener on such a thread. That happens
when a helper thread that has the task's `TaskContext`, e.g. `PipedRDD`'s stdin
writer (`PipedRDD.scala:140`), first reads a broadcast after
`markTaskCompleted` has run the other listeners, while another thread holds the
invocation:
- On a kill thread, before `Executor.scala:916`: `unlock` decrements
`readerCount` (`BlockInfoManager.scala:462`) and then fails with "release lock
on block ... more times than it acquired it"
(`BlockInfoManager.scala:464-465`), since `NON_TASK_WRITER` holds no lock.
`releaseAllLocksForTask(taskId)` then decrements the same lock again
(`BlockInfoManager.scala:568`), so `readerCount` becomes -1. The assertion that
fails there is thrown out of the executor's `finally`, which skips
`cleanUpAllAllocatedMemory()` (`Executor.scala:917`). The block stays at -1, so
a later writer or `removeBlock` never gets the write lock
(`BlockInfoManager.scala:384`).
- In the post-status-update invocation, after `Executor.scala:916`: the
task's lock is already released, so `unlock` either takes another reader's
count or fails. `runListener` then marks a task that has already reported
`FINISHED` as failed and runs its failure listeners.
On master this listener was dropped, and `releaseAllLocksForTask` released
the lock for the right task. These are `Predef.assert`s, which are not elided
in production. The window is narrow, but the result is broken lock bookkeeping,
not the NPE that item 5 was about. Could the loop set the `TaskContext` to
`this` while it runs listeners left by other threads, when the thread has none?
Passing the context to `releaseLock` would not help: its SPARK-27666 check
skips the release once the task has completed.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -217,30 +240,85 @@ private[spark] class TaskContextImpl(
// writer that adds listeners lazily while it pulls its input. And since
the completion
// listeners only run after `completed` is set, they run after every
pre-completion listener,
// even if another thread is invoking listeners at this point.
+ //
+ // The completion listeners then run on this thread too, and this method
returns only after
+ // they have run. If it returned while another thread is invoking
listeners (e.g. a kill thread
+ // running failure listeners), that thread would run them only after its
own listeners, after
+ // the task has released its memory and reported its result. So the
listener invocation is
+ // claimed together with setting `completed`, which also keeps a thread
that adds a completion
+ // listener at that moment from running them; if another thread holds it,
that thread hands it
+ // over to this one when done. Failure listeners pending at that point run
first (see
+ // `invokeListeners()`). A completion listener added after this method has
run the others runs
+ // on the thread adding it, or, if another thread is invoking listeners at
that moment, on that
+ // thread before it stops (see `invokeListeners()`).
+ //
+ // The wait for the other thread has no time bound: if its listeners
block, so does this task.
+ // It logs a warning periodically, the first one with that thread's stack
trace. The task waits
+ // here before it releases its block locks and memory (the executor does
that once `Task.run`
+ // returns), so a failure listener on another thread, e.g. a kill thread,
must not wait for
+ // either. If such a task is killed, `spark.task.reaper.enabled` with a
positive
Review Comment:
**4. [Low] Nothing bounds the wait for a kill-thread holder**
This now says that a positive `spark.task.reaper.killTimeout` bounds the
wait, "except when the blocked listeners run on the reaper thread itself,
within its kill call". With `spark.task.reaper.enabled`, though,
`Executor.killTask` runs every kill on a reaper thread
(`Executor.scala:615-633`). `TaskReaper.run` calls `taskRunner.kill`
(`Executor.scala:1283`) before its monitoring loop and its timeout check
(`Executor.scala:1289`, `:1322`). A later reaper with `interruptThread = true`
(`Executor.scala:619`) blocks on the `TaskRunner` lock that the first one holds
(`Executor.scala:735`). So if a failure listener blocks on a kill thread, which
is the example here, no reaper ever times out, whatever the configuration. The
reaper bounds the wait only when another kind of thread holds the invocation,
e.g. a helper thread that called `markTaskFailed`, and the task is killed.
Could the comment and the description say so? For example: "Nothing bounds a
wait for a kill thread: with `spark.task.reaper.enabled`, every kill runs on a
reaper thread, which checks its timeout only after its kill call returns. If
another thread holds the invocation and the task is killed, ..." A deeper
option, probably for a separate JIRA, is to have `markInterrupted` only record
the failure and leave the failure listeners to the task thread, so that a kill
thread never holds the invocation when the task completes. This extends
[round-1 item
6](https://github.com/apache/spark/pull/59663#discussion_r4339138527).
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -412,21 +596,62 @@ private[spark] class TaskContextImpl(
// [[markTaskFailed]] here will have no effect and we simply
resume running the
// remaining completion handlers.
try {
- listenerInvocationThread = None
- markTaskFailed(e)
+ markTaskFailed(e, Nested)
Review Comment:
**3. [Low] `runListener` still hides fatal errors from the nested failure
invocation**
Round 2 narrowed `runLeftListeners` (L615), so that anything other than a
listener failure propagates, as my [round-1 item
2](https://github.com/apache/spark/pull/59663#discussion_r4339138480) asked.
The nested failure invocation started on this line still goes through `case t:
Throwable => e.addSuppressed(t)` (L601). Since a `Nested` invocation does not
release, an `OutOfMemoryError` from its own code, e.g. from `logError` (L604)
when a failure listener fails, becomes a suppressed exception of `e`. `e` then
becomes a suppressed exception of the `TaskCompletionListenerException`, which
has no cause, so `Executor.isFatalError` does not see the error and the
executor keeps running. The same error from the nested invocation at L621-623
propagates. `invokeTaskPreCompletionListener` has the same catch (L343-347).
Could L601 and L346 catch only `TaskCompletionListenerException`, like L615?
Alternatively, a plain `markTaskFailed(e)` would only record the failure here,
because this thread holds the invocation. The step at L621-623 would then run
the failure listeners before the next listener, and the private
`markTaskFailed` overload would no longer be needed.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -280,27 +358,115 @@ 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. If the task
+ * body succeeded, it lets a fatal error reach the executor, which then
exits instead of waiting
+ * for that thread; if the task body failed, `runTaskWithListeners` adds the
error to the task's
+ * error as a suppressed exception. And if the other thread handed the
invocation over just
+ * before the failure, `markTaskCompleted()` releases it, so the completion
listeners run only
+ * once another listener is added. `markTaskCompleted()` also removes this
thread as the next
+ * invocation thread when the wait ends this way.
+ */
+ 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 match {
+ case Some(holder) =>
+ val holderStack = if (warned) None else Some(holder.getStackTrace)
+ // If the holder has handed over meanwhile, its stack shows
whatever it does next.
+ if (synchronized(listenerInvocationThread.contains(holder))) {
+ val waitedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNs)
+ val message = log"Task ${MDC(TASK_ATTEMPT_ID, taskAttemptId)}
has been waiting " +
+ log"for ${MDC(TOTAL_TIME, waitedMs)} ms to run its completion
listeners until " +
+ log"thread ${MDC(THREAD_NAME, holder.getName)} finishes
running its task listeners."
+ holderStack match {
+ case Some(trace) =>
+ val stack = new Exception(s"Stack trace of thread
${holder.getName}")
+ stack.setStackTrace(trace)
+ logWarning(message, stack)
Review Comment:
**2. [Low] A non-fatal failure of the handover warning still ends the wait**
The scaladoc at L371-376 justifies ending this wait when logging fails: if
the task body succeeded, a fatal error reaches the executor, which exits. That
does not cover a non-fatal failure, and the new test uses one
(`IllegalStateException`, `TaskContextSuite.scala:1308`). With an appender that
has `ignoreExceptions = false`, this `logWarning` throws an
`AppenderLoggingException`, and then:
- `markTaskCompleted` throws, so `Task.run` fails.
- The executor releases the task's block locks and frees its pages
(`Executor.scala:916-917`), reports the failure and keeps running.
- Once the holder's listener returns, the leftover branch at L627 runs every
completion listener on the holder, on that freed state. For example,
`UnsafeExternalSorter`'s cleanup listener (`UnsafeExternalSorter.java:206`)
frees pages that `cleanUpAllAllocatedMemory` has already freed, so `freePage`
indexes `pageTable` with a negative page number (`TaskMemoryManager.java:804`).
Broadcast release listeners run into item 1.
On master these listeners were dropped instead.
`PipelinedWriterTask.awaitExit` keeps waiting for this reason
(`PythonRunner.scala:390-392`, "a failure here, e.g. an OutOfMemoryError while
logging, must not end the wait"). Could this wait do the same: keep the first
failure, go on waiting, and rethrow it after the handover? This extends
[round-1 item
7](https://github.com/apache/spark/pull/59663#discussion_r4339138538).
##########
core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleSuite.scala:
##########
@@ -1372,68 +1372,81 @@ class StreamingShuffleSuite
}
}
- // Demonstrates why background threads must not call
context.markTaskFailed() directly:
- // doing so can cause the task thread's markTaskCompleted() to silently skip
completion
- // listeners. In production, cleanupResources() is a completion listener, so
skipping it
- // leaks threads. The reader avoids this by routing background errors
through a
- // CompletableFuture so markTaskFailed/markTaskCompleted are both called
from the task thread.
+ // Shows why background threads should not call context.markTaskFailed()
directly: the
+ // failure listeners then run on that thread, and the task thread's
markTaskCompleted() blocks
+ // until they finish before it runs the completion listeners. In production,
cleanupResources()
+ // is a completion listener, so it waits for whatever the background
thread's failure listeners
+ // wait for. The reader avoids this by routing background errors through an
ErrorNotifier so
+ // that markTaskFailed/markTaskCompleted are both called from the task
thread.
test("background markTaskFailed races with task thread markTaskCompleted") {
val readerContext = createTaskContext(sparkConf, partitionId = 0)
TaskContext.setTaskContext(readerContext)
Review Comment:
**5. [Low] A failure in this test leaves `readerContext` set on the
ScalaTest thread**
Since `markTaskCompleted` moved to `taskThread`, which sets its own context
(L1421), nothing on the test thread reads the context set here.
`TaskContext.unset()` (L1444) runs only if every assertion passed. If one
fails, e.g. the `eventually` at L1427-1429 under the regression this test
guards against, the completed and failed `readerContext` stays set on this
thread. The next tests that create a `SparkContext` (L1464, L1506, L1575) then
fail with "SparkContext should only be created and accessed on the driver"
(`SparkContext.scala:3411`, through `Utils.isInRunningSparkTask`), and so can
later suites that run on the same thread. Could L1383 and L1444 be removed?
##########
core/src/main/scala/org/apache/spark/util/ErrorNotifier.scala:
##########
@@ -28,9 +28,10 @@ import org.apache.spark.internal.Logging
*
* A background thread records the first error via [[markError]] rather than
failing the task
* directly; the owning thread polls [[throwErrorIfExists]] (or [[getError]])
at a safe point and
- * re-throws on its own thread. This avoids a race where calling
`TaskContext.markTaskFailed` from
- * a background thread can run concurrently with the task thread's
`markTaskCompleted` and silently
- * skip completion listeners (including cleanup), leading to thread leaks.
+ * re-throws on its own thread. Calling `TaskContext.markTaskFailed` from a
background thread
Review Comment:
**10. [Nit] This holds only if the background thread gets the listener
invocation first**
If the task thread is already running completion listeners, a background
`markTaskFailed` only records the failure and returns
(`TaskContextImpl.scala:546-548`), and the task thread runs the failure
listeners before its next completion listener
(`TaskContextImpl.scala:621-623`). Nothing blocks then. "a failure listener
left to the completing thread runs before the next completion listener" tests
exactly that (`TaskContextSuite.scala:1129-1130`). Maybe "Calling
`TaskContext.markTaskFailed` from a background thread can instead run the
failure listeners on that thread, if no other thread is invoking listeners, and
a concurrent `markTaskCompleted` on the task thread then waits for them before
it runs the completion listeners (including cleanup)."
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -134,7 +149,7 @@ private[spark] class TaskContextImpl(
completed
}
if (needToCallListener) {
- invokeTaskCompletionListeners(None)
+ invokeTaskCompletionListeners(None, Claim)
Review Comment:
**11. [Nit] The comment above this call does not cover the handover**
(L142-144, outside the diff)
"adding the new listener to `onCompleteCallbacks` will cause that thread to
execute the new listener" no longer holds while `markTaskCompleted` waits for
the handover. The holder then skips its leftover branch (L627) and hands the
invocation over, and the waiting thread runs the new listener. "completion
waits for another thread invoking listeners and runs them" asserts this for
"added" (`TaskContextSuite.scala:1075`). Maybe "If another thread is invoking
listeners, the call below is a no-op: that thread runs the new listener before
it stops, or hands it over with the others to the thread waiting in
`markTaskCompleted()`."
##########
core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala:
##########
@@ -931,6 +931,466 @@ 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")
+ }
+
+ /**
+ * Runs `takeInvocation` on a new thread named `name`, by default failing
the task, and waits
+ * until a failure listener on it holds the listener invocation, which it
does until the
+ * returned latch is counted down. Returns that latch and the thread.
+ */
+ private def holdListenerInvocation(context: TaskContext, name: String =
"listener-holder")(
+ takeInvocation: => Unit = context.markTaskFailed(new Exception("failed
on another thread")))
+ : (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(name)(takeInvocation)
+ assert(holding.await(1, TimeUnit.MINUTES))
+ (release, holder)
+ }
+
+ /**
+ * Fails every logging call whose message contains `trigger` with `failure`.
Unlike
+ * `LogAppender`, it does not ignore its exceptions, so log4j passes them to
the caller.
+ */
+ private class FailingAppender(trigger: String, failure: RuntimeException)
+ extends AbstractAppender("failingAppender", null, null, false,
Property.EMPTY_ARRAY) {
+ override def append(event: LogEvent): Unit = {
+ if (event.getMessage.getFormattedMessage.contains(trigger)) {
+ throw failure
+ }
+ }
+ }
+
+ private def isCausedBy(error: Throwable, cause: Throwable): Boolean =
+ Iterator.iterate(error)(_.getCause).takeWhile(_ != null).exists(_ eq cause)
+
+ /** Waits until `thread` waits in `markTaskCompleted()` for the listener
invocation. */
+ private def awaitHandoverWait(thread: Thread): Unit = {
+ eventually(timeout(1.minute)) {
Review Comment:
**12. [Nit] These polls wait a full minute after the thread has returned**
`eventually` keeps checking `getState == Thread.State.TIMED_WAITING` on a
thread that has already terminated. So the regression these tests target, a
`markTaskCompleted` that returns instead of waiting, is reported only after a
minute per test, about five minutes for a full revert. The same holds for
L1236-1239, L1261-1263 and `StreamingShuffleSuite.scala:1427-1429`. Every
holder stays blocked until after the poll, so this would keep what the tests
check:
```scala
eventually(timeout(1.minute)) {
assert(!thread.isAlive || thread.getState ==
Thread.State.TIMED_WAITING)
}
assert(thread.isAlive, s"${thread.getName} returned instead of waiting")
```
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -217,30 +240,85 @@ private[spark] class TaskContextImpl(
// writer that adds listeners lazily while it pulls its input. And since
the completion
// listeners only run after `completed` is set, they run after every
pre-completion listener,
// even if another thread is invoking listeners at this point.
+ //
+ // The completion listeners then run on this thread too, and this method
returns only after
+ // they have run. If it returned while another thread is invoking
listeners (e.g. a kill thread
+ // running failure listeners), that thread would run them only after its
own listeners, after
+ // the task has released its memory and reported its result. So the
listener invocation is
Review Comment:
**7. [Nit] Two statements here do not hold when the holder is a kill thread**
In [round-1 item
8](https://github.com/apache/spark/pull/59663#discussion_r4339138551) I
suggested "after the task has released its memory and reported its result".
That is wrong for the kill thread this sentence names. It runs the failure
listeners inside `TaskRunner.kill`'s `synchronized` (`Executor.scala:735-737`),
and every path to a status update first calls
`setTaskFinishedAndClearInterruptStatus` (`Executor.scala:746`), which needs
that lock. So the task could release its memory before the kill thread is done,
but not report its result. Likewise, "which then exits instead of waiting for
that thread" (L375-376) does not hold for a kill thread: the fatal-error path
goes through `setTaskFinishedAndClearInterruptStatus` too
(`Executor.scala:1119`, `:781`), before it reaches the uncaught-exception
handler (`Executor.scala:1145`). Sorry for my wording. Maybe "..., possibly
after the task has released its memory and, unless that thread is a kill
thread, which holds the `TaskRunner
` lock that the status update needs, after it has reported its result"?
##########
core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala:
##########
@@ -216,23 +222,22 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
})
// Interrupt the task from a separate thread and wait until the interrupt
listener starts.
- val interruptThread = new Thread(() => context.markInterrupted("test
interrupt"))
- interruptThread.start()
- interruptListenerStarted.acquire()
-
- // While the interrupt listener is running on the interrupt thread, mark
the task completed
- // on this thread. With the dedicated interrupt listener loop, this must
NOT be blocked.
- context.markTaskCompleted(None)
-
- // The completion listener should have been called even though the
interrupt listener is still
- // running. If `invokeListeners()` were shared between interrupt and
completion listeners,
- // the completion listener would be silently skipped because
`listenerInvocationThread` would
- // be held by the interrupt thread.
- verify(completionListener, times(1)).onTaskCompletion(any())
-
- // Release the interrupt listener and join the interrupt thread.
- interruptListenerRelease.release()
- interruptThread.join()
+ val interruptThread =
startDaemonThread("interrupt")(context.markInterrupted("test interrupt"))
+ try {
+ interruptListenerStarted.acquire()
+
+ // While the interrupt listener is running on the interrupt thread, mark
the task completed
+ // on another thread. With the dedicated interrupt listener loop, this
must NOT be blocked.
+ // If `invokeListeners()` were shared between interrupt and completion
listeners, the
+ // interrupt thread would hold `listenerInvocationThread`, and the
completion would wait
+ // for it, so the join below would time out.
+
joinThread(startDaemonThread("completion")(context.markTaskCompleted(None)))
Review Comment:
**8. [Nit] The rewritten tests no longer fail when `markTaskCompleted`
throws**
The base version of this test called `markTaskCompleted` on the test thread,
as did the race test in `StreamingShuffleSuite` (now
`StreamingShuffleSuite.scala:1420-1423`), so an exception from it failed the
test. On a daemon thread it only reaches the default uncaught-exception
handler: `joinThread` checks only that the thread has ended, and `verify(...,
times(1))` only that the listener ran. For example, if the release moved out of
the `synchronized` in `nextStep()`, `notifyAll()` would throw
`IllegalMonitorStateException` after all listeners ran, and both tests would
still pass. Could `startDaemonThread` record what its body throws, and
`joinThread` assert that it is null? That would also cover the
`markTaskCompleted` threads at L1060, L1151, L1260 and L1357.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -331,10 +497,34 @@ private[spark] class TaskContextImpl(
}
}
+ /**
+ * Invokes and removes the listeners in `listeners`, in reverse order of
registration.
+ *
+ * Other threads cannot invoke listeners while this thread does, so they
queue completion and
+ * failure listeners and leave them to this thread. (Post-status-update
listeners that another
+ * thread invokes meanwhile are not left to it; they are dropped.) Failure
listeners left to it
+ * (the task has failed) run before the next listener of `listeners`, e.g.
so that a failure
+ * listener that aborts a state store update runs before the completion
listener that releases
+ * the state store. Unless `mode` is `Nested`, this thread releases the
invocation when it is
+ * done, but first runs the completion listeners left to it if the task has
completed, unless a
+ * thread waiting in `markTaskCompleted()` takes the invocation over and
runs them itself. Such
+ * leftover completion listeners were added after `markTaskCompleted()` had
run the others.
Review Comment:
**9. [Nit] Leftover completion listeners are not always ones added after
`markTaskCompleted()` ran the others**
When the handover wait ends early, `markTaskCompleted` removes itself as the
waiter, and the holder's leftover branch (L627) runs the listeners that
`markTaskCompleted` never ran. L373-374 says so, and "a failing handover
warning does not leave a stale waiter" asserts it
(`TaskContextSuite.scala:1306`, `:1329`). Maybe "Such leftover completion
listeners are usually ones added after `markTaskCompleted()` had run the
others, or all of them if it stopped waiting for the handover."
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -217,30 +240,85 @@ private[spark] class TaskContextImpl(
// writer that adds listeners lazily while it pulls its input. And since
the completion
// listeners only run after `completed` is set, they run after every
pre-completion listener,
// even if another thread is invoking listeners at this point.
+ //
+ // The completion listeners then run on this thread too, and this method
returns only after
+ // they have run. If it returned while another thread is invoking
listeners (e.g. a kill thread
+ // running failure listeners), that thread would run them only after its
own listeners, after
+ // the task has released its memory and reported its result. So the
listener invocation is
+ // claimed together with setting `completed`, which also keeps a thread
that adds a completion
+ // listener at that moment from running them; if another thread holds it,
that thread hands it
+ // over to this one when done. Failure listeners pending at that point run
first (see
+ // `invokeListeners()`). A completion listener added after this method has
run the others runs
+ // on the thread adding it, or, if another thread is invoking listeners at
that moment, on that
+ // thread before it stops (see `invokeListeners()`).
+ //
+ // The wait for the other thread has no time bound: if its listeners
block, so does this task.
+ // It logs a warning periodically, the first one with that thread's stack
trace. The task waits
+ // here before it releases its block locks and memory (the executor does
that once `Task.run`
+ // returns), so a failure listener on another thread, e.g. a kill thread,
must not wait for
+ // either. If such a task is killed, `spark.task.reaper.enabled` with a
positive
+ // `spark.task.reaper.killTimeout` bounds the wait by killing the executor
JVM, except in
+ // local mode, where the reaper only logs an error, and except when the
blocked listeners run
+ // on the reaper thread itself, within its kill call.
+ val currentThread = Thread.currentThread()
+ var mode: InvocationMode = Claimed
+ var awaitHandover = false
+ // Whether this call registered this thread as the listener invocation
thread or as the next
+ // one, so that it must make sure not to leave it registered.
+ var registered = false
val preCompletionExceptions = new ArrayBuffer[Throwable](2)
def nextPreCompletionListener(): Option[TaskCompletionListener] =
synchronized {
if (onPreCompleteCallbacks.empty()) {
// Set together with finding no more pre-completion listeners, so that
one added later
- // is handed to `addTaskCompletionListener` instead of being left in
the stack.
+ // is handed to `addTaskCompletionListener` instead of being left in
the stack. The
+ // listener invocation is claimed in the same step (see above).
completed = true
+ if (listenerInvocationThread.isEmpty) {
+ listenerInvocationThread = Some(currentThread)
+ registered = true
+ } else if (listenerInvocationThread.contains(currentThread)) {
+ // Called from a listener on this thread: run the completion
listeners as part of that
+ // invocation.
+ mode = Nested
+ } else {
+ nextListenerInvocationThread = Some(currentThread)
+ awaitHandover = true
+ registered = true
+ }
None
} else {
Some(onPreCompleteCallbacks.pop())
}
}
var listener: Option[TaskCompletionListener] = None
- while ({listener = nextPreCompletionListener(); listener.nonEmpty}) {
- invokeTaskPreCompletionListener(listener.get, preCompletionExceptions)
- }
-
try {
- invokeTaskCompletionListeners(error)
- } catch {
- case e: TaskCompletionListenerException if
preCompletionExceptions.nonEmpty =>
- // Merge the completion listeners' own failures, not the exception
that wraps them. Any
- // other error comes from invokeListeners itself (e.g. an
OutOfMemoryError) and is left
- // to propagate, so that the executor still sees a fatal one.
- preCompletionExceptions ++= e.getSuppressed.toSeq
+ while ({listener = nextPreCompletionListener(); listener.nonEmpty}) {
+ invokeTaskPreCompletionListener(listener.get, preCompletionExceptions)
+ }
+ try {
+ if (awaitHandover) {
+ awaitListenerInvocationHandover(currentThread)
+ }
+ invokeTaskCompletionListeners(error, mode)
+ } catch {
+ case e: TaskCompletionListenerException if
preCompletionExceptions.nonEmpty =>
+ // Merge the completion listeners' own failures, not the exception
that wraps them. Any
+ // other error comes from invokeListeners itself (e.g. an
OutOfMemoryError) and is left
+ // to propagate, so that the executor still sees a fatal one.
+ preCompletionExceptions ++= e.getSuppressed.toSeq
+ }
+ } finally {
+ if (registered) synchronized {
+ // `invokeListeners()` releases the invocation itself once it has
started; this covers an
+ // error before that, or in the wait. A stale waiter would get the
invocation handed to
+ // it, and a stale holder would keep every other thread from running
listeners.
+ if (nextListenerInvocationThread.contains(currentThread)) {
+ nextListenerInvocationThread = None
+ }
+ if (listenerInvocationThread.contains(currentThread)) {
Review Comment:
**6. [Nit] Two of the releases still run in no test**
The reply to [round-1 item
3](https://github.com/apache/spark/pull/59663#discussion_r4339138498) says that
removing either release now fails its test. That holds for the waiter cleanup
at L315-317 and for the release at L646-653 in `Claim` mode. Two branches still
run in no test:
- L318-320. "a failing handover warning does not leave a stale waiter"
releases the holder only after joining the completion thread
(`TaskContextSuite.scala:1314-1324`), so the completing thread is never the
holder when this `finally` runs. "an error that escapes the listener loop"
calls `markTaskCompleted` before the appender fails. Deleting L318-320 fails no
test. Yet without them, a completing thread that got the invocation just before
a failure (the case at L377-379) stays registered for good, and every later
`Claim` returns at L546-548.
- L651 with a waiter. No test lets the holder's loop escape while a thread
waits in `markTaskCompleted`. Replacing `releaseListenerInvocation()` there
with `listenerInvocationThread = None` passes every test, but the waiting task
thread then waits forever, and silently: with no holder, the wait takes `case
None` (L427) and never warns.
Two tests would cover them. In the first, a `FailingAppender` whose `append`
lets the holder hand over (releases it and joins it) and then throws; a
listener added afterwards on the test thread must run there, together with the
pending one. In the second, the holder's failure listener throws and logging
that failure fails while a completion thread waits; the completion thread must
then finish and run the pending listener.
--
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]
