dongjoon-hyun commented on code in PR #59332:
URL: https://github.com/apache/spark/pull/59332#discussion_r4252547607
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -169,37 +193,86 @@ private[spark] class TaskContextImpl(
}
private[spark] override def markTaskFailed(error: Throwable): Unit = {
+ markTaskFailed(error, nested = false)
+ }
+
+ /**
+ * @param nested whether this thread is already invoking listeners, i.e. a
listener failed. The
+ * failure listeners then run as part of that invocation,
which this thread keeps.
+ */
+ private def markTaskFailed(error: Throwable, nested: Boolean): Unit = {
synchronized {
if (failureCauseOpt.isDefined) return
failureCauseOpt = Some(error)
}
- invokeTaskFailureListeners(error)
+ invokeTaskFailureListeners(error, nested)
}
private[spark] override def markTaskCompleted(error: Option[Throwable]):
Unit = {
- synchronized {
+ // The completion listeners always run on this thread, and this method
returns only after they
+ // have run. If any other thread could run them, it could be one the
listeners are meant to
+ // stop and wait for, e.g. a pipelined Python UDF writer that adds
listeners lazily while it
+ // pulls its input, which would then wait for itself. And if this method
returned before they
+ // ran, the task could release its memory while that writer still uses it.
So the listener
+ // invocation is claimed together with setting `completed`; if another
thread holds it (e.g. a
+ // kill thread running failure listeners), that thread hands it over to
this one when done.
+ val nested = synchronized {
if (completed) return
completed = true
+ val currentThread = Thread.currentThread()
+ if (listenerInvocationThread.isEmpty) {
+ listenerInvocationThread = Some(currentThread)
+ false
+ } else if (listenerInvocationThread.contains(currentThread)) {
+ // Called from a listener on this thread: run the completion listeners
as part of that
+ // invocation.
+ true
+ } else {
+ nextListenerInvocationThread = Some(currentThread)
+ var interrupted = false
+ while (!listenerInvocationThread.contains(currentThread)) {
Review Comment:
This wait has no time bound, ignores interrupts, and logs nothing. Before
this PR, `markTaskCompleted` returned right away in this case, so it could not
hang here.
The holder here is typically the kill thread. `Task.kill` runs
`markInterrupted` while holding the global `TaskContext` lock, and with the
task reaper off (the default) `Executor.killTask` runs it on the thread that
received the kill request. If an interrupt listener throws, that thread runs
the failure listeners. If one of them blocks, the task thread stays in `wait()`
forever: kill interrupts are caught at L237 and the loop keeps waiting. Nothing
in the log points at the holder, and with the default
`spark.task.reaper.killTimeout=-1` nothing ever ends the task.
You mentioned this risk in your reply, and I agree it is unlikely. But
`stopAndAwaitExit` already handles the same kind of wait with a 10s warning.
Could this loop do the same, e.g. `wait(timeout)` plus a periodic warning that
names the holder thread (and maybe logs its stack)? Then this case would at
least be diagnosable.
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -183,6 +183,148 @@ private[spark] object BasePythonRunner extends Logging {
ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer")
}
+ /** How often to log while task completion waits for a pipelined writer to
exit. */
+ private val pipelinedWriterExitWarnIntervalMs = 10000L
+
+ /**
+ * Runs `writer` on [[pipelinedWriterThreadPool]] and registers a task
pre-completion listener
+ * that interrupts it and blocks until it has exited.
+ *
+ * Task completion listeners free memory that the writer may still read or
write, e.g. the
+ * pages of the HybridRowQueue that buffers the input rows, or the off-heap
column vectors of a
+ * vectorized reader. If the writer were still running at that point, it
would be a
+ * use-after-free that can crash the executor (SPARK-33277). The listener is
added with
+ * `addTaskPreCompletionListener` so that it runs before all of them,
including the ones that
+ * the writer thread adds lazily while it pulls the upstream iterator, which
would run first if
+ * this were an ordinary completion listener. `Future.get()` cannot be used
for this wait:
+ * once `cancel(true)` succeeds it throws `CancellationException`
immediately, without waiting
+ * for the interrupted runnable to return.
+ *
+ * The wait has no time bound, so a writer blocked in upstream I/O that
ignores interrupts
+ * (e.g. a socket read without a timeout) holds task completion until that
I/O returns. This
+ * includes I/O that only a close listener of the upstream reader would
unblock, since that
+ * listener now runs after the wait. If such a task is killed,
`spark.task.reaper.enabled`
Review Comment:
`spark.task.reaper.enabled` alone does not bound this. With the default
`spark.task.reaper.killTimeout=-1`, `TaskReaper` only monitors the killed task
(and optionally takes thread dumps). `timeoutExceeded()` is false whenever the
timeout is not positive, so the reaper never ends the task.
Could this say something like "`spark.task.reaper.enabled` together with a
positive `spark.task.reaper.killTimeout` bounds how long it may stay in this
state, by killing the executor JVM"? The same applies to the new wait in
`markTaskCompleted`.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -267,13 +352,19 @@ private[spark] class TaskContextImpl(
}
def getNextListenerOrDeregisterThread(): Option[T] = synchronized {
- if (listeners.empty()) {
- // We have executed all listeners that have been added so far.
Deregister this thread as the
- // callback invocation thread.
- listenerInvocationThread = None
- None
- } else {
- Some(listeners.pop())
+ listenerStacks.find(!_.empty()) match {
+ case Some(listeners) =>
+ Some(listeners.pop())
+ case None =>
+ // We have executed all listeners that have been added so far.
Deregister this thread as
+ // the callback invocation thread, unless an outer invocation on
this thread continues. A
+ // thread waiting in `markTaskCompleted()` takes over directly.
+ if (!nested) {
Review Comment:
Failure listeners that another thread queues while this thread holds the
invocation never run. `markTaskFailed` and `addTaskFailureListener` from
another thread find `listenerInvocationThread` set and return (L342-345). This
loop only pops the completion stacks, so it deregisters here and leaves
`onFailureCallbacks` untouched.
This predates the PR, but the window used to be short, because the old
`writerFuture.get()` returned right away. Now the task thread holds the
invocation for the whole wait in `stopAndAwaitExit`. For example, if a kill
arrives during that wait and an interrupt listener throws, `markInterrupted`
records the failure on the kill thread, but no failure listener ever runs.
Would it make sense to run pending failure listeners before deregistering
when `failureCauseOpt` is set? Or at least to note the gap in the scaladoc?
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -183,6 +183,148 @@ private[spark] object BasePythonRunner extends Logging {
ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer")
}
+ /** How often to log while task completion waits for a pipelined writer to
exit. */
+ private val pipelinedWriterExitWarnIntervalMs = 10000L
+
+ /**
+ * Runs `writer` on [[pipelinedWriterThreadPool]] and registers a task
pre-completion listener
+ * that interrupts it and blocks until it has exited.
+ *
+ * Task completion listeners free memory that the writer may still read or
write, e.g. the
+ * pages of the HybridRowQueue that buffers the input rows, or the off-heap
column vectors of a
+ * vectorized reader. If the writer were still running at that point, it
would be a
+ * use-after-free that can crash the executor (SPARK-33277). The listener is
added with
+ * `addTaskPreCompletionListener` so that it runs before all of them,
including the ones that
+ * the writer thread adds lazily while it pulls the upstream iterator, which
would run first if
+ * this were an ordinary completion listener. `Future.get()` cannot be used
for this wait:
+ * once `cancel(true)` succeeds it throws `CancellationException`
immediately, without waiting
+ * for the interrupted runnable to return.
+ *
+ * The wait has no time bound, so a writer blocked in upstream I/O that
ignores interrupts
+ * (e.g. a socket read without a timeout) holds task completion until that
I/O returns. This
+ * includes I/O that only a close listener of the upstream reader would
unblock, since that
+ * listener now runs after the wait. If such a task is killed,
`spark.task.reaper.enabled`
+ * bounds how long it may stay in this state.
+ */
+ private[python] def startPipelinedWriter(writer: Runnable, context:
TaskContext): Unit = {
+ val writerTask = new PipelinedWriterTask(writer, taskIdentifier(context))
+ val writerFuture = pipelinedWriterThreadPool.submit(writerTask)
+ context.addTaskPreCompletionListener(new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit =
+ writerTask.stopAndAwaitExit(writerFuture)
+ })
+ }
+
+ /**
+ * Wraps a pipelined writer so that task completion can wait until the
writer has either
+ * exited or is guaranteed never to run.
+ */
+ private[python] class PipelinedWriterTask(
+ writer: Runnable,
+ taskName: String,
+ warnIntervalMs: Long = pipelinedWriterExitWarnIntervalMs)
+ extends Runnable {
+ // Set by whichever of run() and stopAndAwaitExit() gets here first. If
run() wins, the
+ // writer runs and stopAndAwaitExit() waits for it; otherwise run() skips
the writer.
+ private val claimed = new AtomicBoolean(false)
Review Comment:
nit: `claimed` is set by whichever side gets there first, so afterwards it
cannot tell "the writer is running" from "the stopper cancelled it". A second
`stopAndAwaitExit` on a writer that the first call stopped before it started
would fail the CAS, skip the early return, and wait on `exited` forever,
because `run()` never counts it down when it loses the CAS.
Only one listener calls this today, so it cannot happen yet. One state field
guarded by `this` (not started / `Running(thread)` / done or cancelled), with
the latch counted down on every terminal transition, would remove this case and
also the split between the CAS and the `synchronized` `writerThread`.
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -183,6 +183,148 @@ private[spark] object BasePythonRunner extends Logging {
ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer")
}
+ /** How often to log while task completion waits for a pipelined writer to
exit. */
+ private val pipelinedWriterExitWarnIntervalMs = 10000L
+
+ /**
+ * Runs `writer` on [[pipelinedWriterThreadPool]] and registers a task
pre-completion listener
+ * that interrupts it and blocks until it has exited.
+ *
+ * Task completion listeners free memory that the writer may still read or
write, e.g. the
+ * pages of the HybridRowQueue that buffers the input rows, or the off-heap
column vectors of a
+ * vectorized reader. If the writer were still running at that point, it
would be a
+ * use-after-free that can crash the executor (SPARK-33277). The listener is
added with
+ * `addTaskPreCompletionListener` so that it runs before all of them,
including the ones that
+ * the writer thread adds lazily while it pulls the upstream iterator, which
would run first if
+ * this were an ordinary completion listener. `Future.get()` cannot be used
for this wait:
+ * once `cancel(true)` succeeds it throws `CancellationException`
immediately, without waiting
+ * for the interrupted runnable to return.
+ *
+ * The wait has no time bound, so a writer blocked in upstream I/O that
ignores interrupts
+ * (e.g. a socket read without a timeout) holds task completion until that
I/O returns. This
+ * includes I/O that only a close listener of the upstream reader would
unblock, since that
+ * listener now runs after the wait. If such a task is killed,
`spark.task.reaper.enabled`
+ * bounds how long it may stay in this state.
+ */
+ private[python] def startPipelinedWriter(writer: Runnable, context:
TaskContext): Unit = {
+ val writerTask = new PipelinedWriterTask(writer, taskIdentifier(context))
+ val writerFuture = pipelinedWriterThreadPool.submit(writerTask)
+ context.addTaskPreCompletionListener(new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit =
+ writerTask.stopAndAwaitExit(writerFuture)
+ })
+ }
+
+ /**
+ * Wraps a pipelined writer so that task completion can wait until the
writer has either
+ * exited or is guaranteed never to run.
+ */
+ private[python] class PipelinedWriterTask(
+ writer: Runnable,
+ taskName: String,
+ warnIntervalMs: Long = pipelinedWriterExitWarnIntervalMs)
+ extends Runnable {
+ // Set by whichever of run() and stopAndAwaitExit() gets here first. If
run() wins, the
+ // writer runs and stopAndAwaitExit() waits for it; otherwise run() skips
the writer.
+ private val claimed = new AtomicBoolean(false)
+ private val exited = new CountDownLatch(1)
+ // The thread running the writer, while it does. Protected by `this`, so
that it is only
+ // interrupted while it still runs the writer, not after it has moved on
to other work.
+ private var writerThread: Thread = _
+
+ override def run(): Unit = {
+ if (claimed.compareAndSet(false, true)) {
+ synchronized {
+ writerThread = Thread.currentThread()
+ }
+ try {
+ writer.run()
+ } catch {
+ case t: Throwable =>
+ // The thread pool's FutureTask would keep this to itself, and the
result of a
+ // cancelled future cannot be retrieved, so log it here.
+ logError(log"Pipelined Python writer of ${MDC(TASK_NAME,
taskName)} failed", t)
+ throw t
+ } finally {
+ synchronized {
+ writerThread = null
+ }
+ exited.countDown()
+ }
+ }
+ }
+
+ private def isWriterThread: Boolean = synchronized {
+ writerThread eq Thread.currentThread()
+ }
+
+ private def interruptWriter(): Unit = synchronized {
+ if (writerThread != null) {
+ writerThread.interrupt()
+ }
+ }
+
+ def stopAndAwaitExit(future: Future[_]): Unit = {
+ // Interrupts the writer thread if the writer is running. This unblocks
channel.write
+ // (the JDK closes the channel and throws ClosedByInterruptException).
Otherwise the writer
+ // sees the interrupt only between calls to writeNextInputToStream, each
of which can pull
+ // and serialize a whole batch of input rows (e.g. for the Arrow
runners).
+ future.cancel(true)
+ if (claimed.compareAndSet(false, true)) {
+ // The writer never started and now never will.
+ return
+ }
+ if (isWriterThread) {
Review Comment:
I don't think "The interrupt above stops the writer afterwards" holds. On
this thread, `cancel(true)` interrupts the `FutureTask` runner, which is this
thread itself, so all it does is set our own interrupt flag. After this
listener returns:
- The remaining completion listeners run on the writer thread with that flag
set. The comment at L296-299 says this makes their blocking channel and lock
operations fail and skip cleanup.
- Those listeners close the HybridRowQueue pages and the upstream reader's
off-heap vectors. When they return, control goes back to the upstream code that
triggered them (e.g. `ParquetReaderCallback.initIfNotAlready`). That code keeps
using the closed reader, and the writer adds the next row to the freed queue
page before the loop at L1445 checks `isInterrupted`.
So when this path is taken, it trades the hang for the SPARK-33277
use-after-free, and only logs a WARN. I agree it should not be reachable with
`TaskContextImpl` now. Since reaching it means that invariant is broken, could
it log at ERROR, clear the self-interrupt (`Thread.interrupted()`) before
returning, and have the comment say that the writer is *not* stopped before the
remaining listeners run?
##########
core/src/test/scala/org/apache/spark/api/python/BasePythonRunnerSuite.scala:
##########
@@ -50,6 +62,355 @@ class BasePythonRunnerSuite extends SparkFunSuite {
}
}
+ /**
+ * A writer that, once started, runs `onStart` and then keeps running until
`release` is
+ * counted down, ignoring interrupts the way a writer in the middle of
copying a row into a
+ * HybridRowQueue page does.
+ */
+ private class UninterruptibleWriter(
+ onStart: UninterruptibleWriter => Unit = _ => ()) extends Runnable {
+ val started = new CountDownLatch(1)
+ val release = new CountDownLatch(1)
+ @volatile var exited = false
+
+ override def run(): Unit = {
+ try {
+ onStart(this)
+ started.countDown()
+ awaitIgnoringInterrupts(release)
+ } finally {
+ exited = true
+ }
+ }
+ }
+
+ /** Waits for `latch`, ignoring interrupts like a writer in the middle of a
row copy. */
+ private def awaitIgnoringInterrupts(latch: CountDownLatch): Unit = {
+ var done = false
+ while (!done) {
+ try {
+ done = latch.await(1, TimeUnit.MINUTES)
+ } catch {
+ case _: InterruptedException => // keep going
+ }
+ }
+ }
+
+ /** Adds a completion listener that records whether `writer` had exited when
it ran. */
+ private def addCleanupListener(
+ context: TaskContext,
+ writer: UninterruptibleWriter): AtomicReference[Option[Boolean]] = {
+ val exitedBeforeCleanup = new AtomicReference[Option[Boolean]](None)
+ context.addTaskCompletionListener[Unit] { _ =>
exitedBeforeCleanup.set(Some(writer.exited)) }
+ exitedBeforeCleanup
+ }
+
+ /**
+ * Adds a pre-completion listener that runs right before the one added by
+ * `startPipelinedWriter`, so that the test knows task completion has
reached the writer.
+ */
+ private def addReachedListener(context: TaskContext): CountDownLatch = {
+ val reached = new CountDownLatch(1)
+ context.addTaskPreCompletionListener(new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit =
reached.countDown()
+ })
+ reached
+ }
+
+ private def startThread(body: => Unit): Thread = {
+ val thread = new Thread(() => body)
+ thread.setDaemon(true)
+ thread.start()
+ thread
+ }
+
+ test("SPARK-60113: task completion waits for the pipelined writer to exit") {
+ val context = TaskContext.empty()
+ @volatile var cleanupAddedByWriter: AtomicReference[Option[Boolean]] = null
+ val writer = new UninterruptibleWriter(self => {
+ // Like an upstream reader that adds its close listener when the writer
first pulls it.
+ cleanupAddedByWriter = addCleanupListener(context, self)
+ })
+ // Like the listener that closes the HybridRowQueue, added on the task
thread before the
+ // writer starts.
+ val cleanupAddedBefore = addCleanupListener(context, writer)
+ BasePythonRunner.startPipelinedWriter(writer, context)
+ assert(writer.started.await(1, TimeUnit.MINUTES))
+ val reached = addReachedListener(context)
+
+ val completion = startThread(context.markTaskCompleted(None))
+ try {
+ assert(reached.await(1, TimeUnit.MINUTES))
+ // Task completion must stay blocked while the writer is still running.
+ completion.join(500)
+ assert(completion.isAlive)
+ assert(cleanupAddedBefore.get.isEmpty)
+ assert(cleanupAddedByWriter.get.isEmpty)
+ } finally {
+ writer.release.countDown()
+ }
+ completion.join(TimeUnit.MINUTES.toMillis(1))
+ assert(!completion.isAlive)
+ assert(cleanupAddedBefore.get === Some(true))
+ assert(cleanupAddedByWriter.get === Some(true))
+ }
+
+ test("SPARK-60113: a listener the writer adds once completion has started
waits for it") {
+ val context = TaskContext.empty()
+ val addNow = new CountDownLatch(1)
+ val added = new CountDownLatch(1)
+ // Whether the writer had exited, and the thread, when the listener added
by the writer ran.
+ val lateCleanup = new AtomicReference[Option[(Boolean, Thread)]](None)
+ val writerStarted = new CountDownLatch(1)
+ val release = new CountDownLatch(1)
+ // Like a writer that adds a listener when it pulls more input after
completion has started.
+ val lateWriter: Runnable = new Runnable {
+ @volatile var exited = false
+ override def run(): Unit = {
+ try {
+ writerStarted.countDown()
+ awaitIgnoringInterrupts(addNow)
+ context.addTaskCompletionListener[Unit] { _ =>
+ lateCleanup.set(Some((exited, Thread.currentThread())))
+ }
+ added.countDown()
+ awaitIgnoringInterrupts(release)
+ } finally {
+ exited = true
+ }
+ }
+ }
+ BasePythonRunner.startPipelinedWriter(lateWriter, context)
+ assert(writerStarted.await(1, TimeUnit.MINUTES))
+ val reached = addReachedListener(context)
+
+ val completion = startThread(context.markTaskCompleted(None))
+ try {
+ assert(reached.await(1, TimeUnit.MINUTES))
+ addNow.countDown()
+ assert(added.await(1, TimeUnit.MINUTES))
+ // The writer thread only queued the listener: task completion still
waits for the writer.
+ completion.join(500)
+ assert(completion.isAlive)
+ assert(lateCleanup.get.isEmpty)
+ } finally {
+ release.countDown()
+ }
+ completion.join(TimeUnit.MINUTES.toMillis(1))
+ assert(!completion.isAlive)
+ assert(lateCleanup.get === Some((true, completion)))
+ }
+
+ test("SPARK-60113: an interrupt does not stop the wait and is not passed to
later listeners") {
+ val context = TaskContext.empty()
+ val writer = new UninterruptibleWriter
+ val cleanup = addCleanupListener(context, writer)
+ val interruptedInCleanup = new AtomicReference[Option[Boolean]](None)
+ context.addTaskCompletionListener[Unit] { _ =>
+ interruptedInCleanup.set(Some(Thread.currentThread().isInterrupted))
+ }
+ BasePythonRunner.startPipelinedWriter(writer, context)
+ assert(writer.started.await(1, TimeUnit.MINUTES))
+ val reached = addReachedListener(context)
+
+ val completion = startThread(context.markTaskCompleted(None))
+ try {
+ assert(reached.await(1, TimeUnit.MINUTES))
+ // Give the completion thread time to block in the wait, then interrupt
it, e.g. a task
+ // kill that lands while the listeners run.
+ completion.join(200)
Review Comment:
Since `1053b61d411` keeps an interrupt that is pending when the wait starts,
this test now depends on timing. The completion thread may not reach
`Thread.interrupted()` in `stopAndAwaitExit` within these 200 ms. It can still
be between `reached.countDown()` and the stop listener, e.g. after a GC pause
on a busy CI machine. The interrupt then counts as pending, the `finally`
restores it, and `interruptedInCleanup` ends up `Some(true)`, so the assertion
at L232 fails.
Could this wait until the thread blocks in the wait instead, as the other
new tests do?
```scala
eventually(timeout(1.minute)) {
assert(completion.getState == Thread.State.TIMED_WAITING)
}
completion.interrupt()
```
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -239,10 +312,22 @@ private[spark] class TaskContextImpl(
}
}
+ /**
+ * Invokes and removes the listeners in `listenerStacks`, each in reverse
order of registration.
+ * A listener is always taken from the first non-empty stack, so all
listeners of a stack run
+ * before those of the stacks after it that are still pending.
+ *
+ * @param claimed whether the caller has already registered this thread as
+ * `listenerInvocationThread`.
+ * @param nested whether that registration belongs to an outer invocation on
this thread, which
+ * keeps it when this one is done. Implies `claimed`.
+ */
private def invokeListeners[T](
- listeners: Stack[T],
+ listenerStacks: Seq[Stack[T]],
name: String,
- error: Option[Throwable])(
+ error: Option[Throwable],
+ claimed: Boolean = false,
Review Comment:
nit: `nested` implies `claimed`, but the two booleans still allow `nested =
true, claimed = false`. That combination would register the thread at L350 and
then never deregister it, because the `nested` branch at L362 skips that.
`listenerInvocationThread` would stay set, and a later `markTaskCompleted` from
another thread would wait forever.
Every caller passes a valid combination today (`(nested, nested)` and
`(true, nested)`), so this is not a bug. A single three-state parameter (e.g.
claim / already claimed / nested) would make the invalid case unrepresentable,
and `invokeTaskCompletionListeners` would not need its own defaults.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -169,37 +193,86 @@ private[spark] class TaskContextImpl(
}
private[spark] override def markTaskFailed(error: Throwable): Unit = {
+ markTaskFailed(error, nested = false)
+ }
+
+ /**
+ * @param nested whether this thread is already invoking listeners, i.e. a
listener failed. The
+ * failure listeners then run as part of that invocation,
which this thread keeps.
+ */
+ private def markTaskFailed(error: Throwable, nested: Boolean): Unit = {
synchronized {
if (failureCauseOpt.isDefined) return
failureCauseOpt = Some(error)
}
- invokeTaskFailureListeners(error)
+ invokeTaskFailureListeners(error, nested)
}
private[spark] override def markTaskCompleted(error: Option[Throwable]):
Unit = {
- synchronized {
+ // The completion listeners always run on this thread, and this method
returns only after they
+ // have run. If any other thread could run them, it could be one the
listeners are meant to
+ // stop and wait for, e.g. a pipelined Python UDF writer that adds
listeners lazily while it
+ // pulls its input, which would then wait for itself. And if this method
returned before they
+ // ran, the task could release its memory while that writer still uses it.
So the listener
+ // invocation is claimed together with setting `completed`; if another
thread holds it (e.g. a
+ // kill thread running failure listeners), that thread hands it over to
this one when done.
Review Comment:
With this change, the class doc of `ErrorNotifier` (L31-33) is out of date.
It says that calling `markTaskFailed` from a background thread "can run
concurrently with the task thread's `markTaskCompleted` and silently skip
completion listeners". Now `markTaskCompleted` waits for that thread and then
runs them.
The hazards are different now: the task thread blocks until the background
thread's failure listeners finish, and failure listeners that another thread
queues during completion are dropped (see the comment on L362). The comment in
`StreamingShuffleReader.checkTaskFailure` ("run without contention") is in the
same situation. Could both be updated here, since this PR changes the behavior
they describe?
--
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]
