[SPARK-24309][CORE] AsyncEventQueue should stop on interrupt. #21356

squito · 2018-05-17T20:49:53Z

EventListeners can interrupt the event queue thread. In particular,
when the EventLoggingListener writes to hdfs, hdfs can interrupt the
thread. When there is an interrupt, the queue should be removed and stop
accepting any more events. Before this change, the queue would continue
to take more events (till it was full), and then would not stop when the
application was complete because the PoisonPill couldn't be added.

Added a unit test which failed before this change.

EventListeners can interrupt the event queue thread. In particular, when the EventLoggingListener writes to hdfs, hdfs can interrupt the thread. When there is an interrupt, the queue should be removed and stop accepting any more events. Before this change, the queue would continue to take more events (till it was full), and then would not stop when the application was complete because the PoisonPill couldn't be added. Added a unit test which failed before this change.

vanzin · 2018-05-17T21:01:26Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

@@ -34,7 +34,8 @@ import org.apache.spark.util.Utils
 * Delivery will only begin when the `start()` method is called. The `stop()` method should be
 * called when no more events need to be delivered.
 */
-private class AsyncEventQueue(val name: String, conf: SparkConf, metrics: LiveListenerBusMetrics)
+private class AsyncEventQueue(val name: String, conf: SparkConf, metrics: LiveListenerBusMetrics,


one param per line now that they don't fit

vanzin · 2018-05-17T21:02:46Z

core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala

+
+    bus.start(mockSparkContext, mockMetricsSystem)
+
+    // after we post one event, the shared queue should get stopped because of the interrupt


s/get stopped/stop

vanzin · 2018-05-17T21:04:36Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

@@ -97,6 +98,11 @@ private class AsyncEventQueue(val name: String, conf: SparkConf, metrics: LiveLi
    } catch {
      case ie: InterruptedException =>
        logInfo(s"Stopping listener queue $name.", ie)
+        stopped.set(true)


I wonder if this is the right thing to do. What would the old bus do here?

I remember there's code to catch exceptions from listeners and just log them, so that other listeners in the same queue are not affected. Not that the event writer queue would suffer from that, but others might.

The old bus would stop the entire spark context (details below).

I dunno what the right behavior is either -- I figured this was the intention given the logInfo. Alternatively we could (a) stop the entire spark context, (b) skip this particular event and keep going or (c) stop the one listener which happened to be active on the interrupt, but keep the queue active (if there were more listeners).

more details on the 2.2 behavior:

ListenerBus.postToAll wouldn't catch the event.

And the polling thread in LiveListenerBus wraps everything in Utils.tryOrStopSparkContext.

I did a similar test on branch-2.2: squito@72951bd

18/05/17 21:38:23.446 SparkListenerBus ERROR Utils: uncaught error in thread SparkListenerBus, stopping SparkContext java.lang.InterruptedException at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1302) at java.util.concurrent.Semaphore.acquire(Semaphore.java:312) at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(LiveListenerBus.scala:80) at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(LiveListenerBus.scala:79) at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(LiveListenerBus.scala:79) at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58) at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1.apply$mcV$sp(LiveListenerBus.scala:78) at org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1282) at org.apache.spark.scheduler.LiveListenerBus$$anon$1.run(LiveListenerBus.scala:77) 18/05/17 21:38:23.448 SparkListenerBus ERROR Utils: throw uncaught fatal error in thread SparkListenerBus java.lang.InterruptedException at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1302) at java.util.concurrent.Semaphore.acquire(Semaphore.java:312) at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(LiveListenerBus.scala:80) at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(LiveListenerBus.scala:79) at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(LiveListenerBus.scala:79) at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58) at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1.apply$mcV$sp(LiveListenerBus.scala:78) at org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1282) at org.apache.spark.scheduler.LiveListenerBus$$anon$1.run(LiveListenerBus.scala:77)

SparkQA · 2018-05-18T01:16:00Z

Test build #90753 has finished for PR 21356 at commit a689f52.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2018-05-18T03:18:41Z

does this mean a problematic listener can kill the queue and crash other listeners in the same queue? Shall we do some isolation?

Not introduced by this PR, but I'm wondering why we only catch InterruptedException? what about other exceptions?

cloud-fan · 2018-05-18T03:21:10Z

also cc @JoshRosen @zsxwing

squito · 2018-05-18T03:29:30Z

does this mean a problematic listener can kill the queue and crash other listeners in the same queue? Shall we do some isolation?

yeah I think marcelo was asking about this above #21356 (comment)

of course, we can't stop a listener from causing an OOM or launching 20 threads and hogging the CPU, but we could try for a little more isolation ...

Not introduced by this PR, but I'm wondering why we only catch InterruptedException? what about other exceptions?

NonFatal is already handled inside ListenerBus.postToAll

SparkQA · 2018-05-18T07:05:02Z

Test build #90769 has finished for PR 21356 at commit 0a44c06.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2018-05-18T07:49:59Z

Since InterruptedException is fatal, maybe it's not safe to keep calling this listener, but stopping the entire queue seems an overkill. Shall we just remove this listener from the queue?

skonto · 2018-05-18T12:43:16Z

As an alternative design option, PoisonPill could be handled differently, since some msgs should have higher priority and can be considered them as part of your "control plane".

squito · 2018-05-18T15:10:45Z

I pushed an update which only removes the listener which was active at the interrupt. Note that is not the same thing as the listener which caused the interrupt, necessarily -- we have no idea who caused the interrupt really, but I guess it would be rather contrived for another Thread to interrupt the event queue threads.

I'm not sure if I should leave some more guards against interrupts as well -- eg. check the interrupted status before posting the event to a specific listener, and if its set throwing InterruptedException, and leaving the generic handling of InterruptedException in the AsyncEventQueue, to shut the queue down entirely if there is any interrupt outside of a posting to a listener.

squito · 2018-05-18T15:11:48Z

As an alternative design option, PoisonPill could be handled differently, since some msgs should have higher priority and can be considered them as part of your "control plane".

I'm pretty sure the existing use of PoisonPill is intentionally put at the end of the queue, so that the everything is processed before the queue is shut down.

skonto · 2018-05-18T16:32:41Z

I'm pretty sure the existing use of PoisonPill is intentionally put at the end of the queue, so that the > everything is processed before the queue is shut down.

Yes I guessed that, I was just thinking that knowing that its to time shut down is one thing and consuming all input is another, anyway.

zsxwing · 2018-05-18T17:10:05Z

hdfs can interrupt the thread.

@squito Could you clarify this? This sounds like a bug in HDFS. I'm curious when this will happen. Although we don't interrupt the AsyncEventQueue thread in Spark right now, I think it's still better to respect InterruptedException rather than ignoring it, e.g., we may decide to interrupt the thread in future.

vanzin · 2018-05-18T18:00:10Z

So, to clarify a little more what the previous code did: ListenerBus.postToAll() catches non-fatal exceptions from each individual listener and ignores them; but InterruptedException is not one of those, so the old code would stop the context like Imran says.

If we just wanted to emulate the old behavior we could do that, although I agree it's better to try not to do that.

It's weird that HDFS would be throwing that exception in the caller thread, but at the same time, it's an issue that we have to work around, since even if they fix HDFS, there are just so many versions out there for that to not matter much.

zsxwing · 2018-05-18T18:10:00Z

@vanzin I just want to understand the real issue here to make sure the current solution is correct.

vanzin · 2018-05-18T18:02:18Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

@@ -187,6 +190,12 @@ private class AsyncEventQueue(val name: String, conf: SparkConf, metrics: LiveLi
    true
  }

+  override def removeListenerOnError(listener: SparkListenerInterface): Unit = {
+    // the listener failed in an unrecoverably way, we want to remove it from the entire
+    // LiveListenerBus (potentially stopping a queue if its empty)


vanzin · 2018-05-18T18:02:45Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

@@ -111,6 +111,12 @@ private[spark] class LiveListenerBus(conf: SparkConf) {
    }
  }

+  private[scheduler] def removeQueue(queue: String): Unit = synchronized {


Not used anymore.

can we remove it?

vanzin · 2018-05-18T18:06:57Z

core/src/main/scala/org/apache/spark/util/ListenerBus.scala

@@ -80,6 +89,11 @@ private[spark] trait ListenerBus[L <: AnyRef, E] extends Logging {
      }
      try {
        doPostEvent(listener, event)
+        if (Thread.interrupted()) {


This is ok right now since Spark code never explicitly interrupts these threads. If we ever need to do that, though, this might become a problem... but in that case I don't know how you'd handle this issue without just giving up and stopping everything.

But... is this correct? Your test just calls Thread.interrupt(), which just sets the interrupt flag. But what if the listener calls an operation that would actually turn that into an InterrruptedException? You're not catching that here anywhere.

scala> Thread.currentThread().interrupt() scala> Thread.sleep(10) java.lang.InterruptedException: sleep interrupted at java.lang.Thread.sleep(Native Method) ... 33 elided

yeah agree, I will handle this too -- but I'll wait to update till there is agreement on the right overall approach

This is ok right now since Spark code never explicitly interrupts these threads. If we ever need to do that, though, this might become a problem... but in that case I don't know how you'd handle this issue without just giving up and stopping everything.

If spark were to explicitly interrupt, then I think we'd also set some other flag indicating a reason, eg. val requestedQueueStop: AtomicBoolean so it shouldn't be hard to distinguish.

I've pushed an update to handle InterruptedException as well.

vanzin · 2018-05-18T18:18:50Z

@zsxwing Imran knows more since he debugged it, but I see a lot of code like this in Hadoop libraries:

try { ... } catch (InterruptedException ie) {
  // log or something
  Thread.currentThread().interrupt();
}

Which could cause things like this. I don't know what would cause the exception in the first place, but the fact that the library is propagating the interrupt to the caller explains why Spark is seeing it in one of its own calls.

zsxwing · 2018-05-18T18:45:36Z

The code here is just propagating the interrupt state and is not doing anything wrong. @squito do you know what would cause this in the first place?

SparkQA · 2018-05-18T19:28:12Z

Test build #90795 has finished for PR 21356 at commit 4a1f657.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

squito · 2018-05-18T20:27:52Z

the problem was not actually an interrupted exception from the listener, it was that the Thread's state was getting set to interrupted, and then there would be a failure later in queue.take():

18/05/16 17:46:36 INFO scheduler.AsyncEventQueue: Stopping listener queue eventLog.
java.lang.InterruptedException
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchronizer.java:1220)
        at java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:335)
        at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:439)
        at org.apache.spark.scheduler.AsyncEventQueue$$anonfun$org$apache$spark$scheduler$AsyncEventQueue$$dispatch$1.apply(AsyncEventQueue.scala:94)
...

so really the question is what to do if a listener sets the interrupt state or throws an InterruptedException. But from all the discussion its not that clear to me what we should (other than of course preventing the queue from hanging and the applciation from stopping).

vanzin · 2018-05-18T22:06:10Z

I think the behavior in your change (trying to keep things running) is fine. Since the queues are isolated now, there's no risk of bad listeners affecting important ones (like the ones used for dynamic allocation or building the UI) and causing cascading issues, which was a risk in the previous versions.

SparkQA · 2018-05-21T06:54:09Z

Test build #90873 has finished for PR 21356 at commit 09d55af.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2018-05-21T07:46:47Z

core/src/main/scala/org/apache/spark/util/ListenerBus.scala

+        if (Thread.interrupted()) {
+          logError(s"Interrupted while posting to ${Utils.getFormattedClassName(listener)}.  " +
+            s"Removing that listener.")
+          removeListenerOnError(listener)


if the thread state is interrupted, is it still safe to keep this queue running?

Thread.interrupted() also clears the interrupted state. So that alone isn't a problem -- we're basically declaring that we've handled the interrupt and nobody else gets to know about it anymore.

You could just throw InterruptedException here to avoid duplicating the error handling code.

vanzin

Looks good aside from the unused method.

vanzin · 2018-05-21T16:49:01Z

core/src/main/scala/org/apache/spark/util/ListenerBus.scala

+        if (Thread.interrupted()) {
+          logError(s"Interrupted while posting to ${Utils.getFormattedClassName(listener)}.  " +
+            s"Removing that listener.")
+          removeListenerOnError(listener)


You could just throw InterruptedException here to avoid duplicating the error handling code.

vanzin

Looks good pending tests.

vanzin · 2018-05-21T19:55:50Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

-    dispatchThread.join()
+    // this thread might be trying to stop itself as part of error handling -- we can't join
+    // in that case.
+    if (Thread.currentThread() != dispatchThread) {


it's ok to leave this, but this doesn't happen anymore, right?

It does still happen, we need this. We see the interrupt in postToAll, which is in the queue thread. If it fails, we call removeListenerOnError. If that results in the queue being empty, we stop the queue.

"spark-listener-group-eventLog" #20 daemon prio=5 os_prio=31 tid=0x00007f831379e800 nid=0x6303 in Object.wait() [0x0000000129226000] java.lang.Thread.State: WAITING (on object monitor) at java.lang.Object.wait(Native Method) - waiting on <0x000000078047ae28> (a org.apache.spark.scheduler.AsyncEventQueue$$anon$1) at java.lang.Thread.join(Thread.java:1245) - locked <0x000000078047ae28> (a org.apache.spark.scheduler.AsyncEventQueue$$anon$1) at java.lang.Thread.join(Thread.java:1319) at org.apache.spark.scheduler.AsyncEventQueue.stop(AsyncEventQueue.scala:135) at org.apache.spark.scheduler.LiveListenerBus$$anonfun$removeListener$2.apply(LiveListenerBus.scala:123) at org.apache.spark.scheduler.LiveListenerBus$$anonfun$removeListener$2.apply(LiveListenerBus.scala:121) at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) at org.apache.spark.scheduler.LiveListenerBus.removeListener(LiveListenerBus.scala:121) - locked <0x0000000780475fe8> (a org.apache.spark.scheduler.LiveListenerBus) at org.apache.spark.scheduler.AsyncEventQueue.removeListenerOnError(AsyncEventQueue.scala:196) at org.apache.spark.scheduler.AsyncEventQueue.removeListenerOnError(AsyncEventQueue.scala:37) at org.apache.spark.util.ListenerBus$class.postToAll(ListenerBus.scala:101)

SparkQA · 2018-05-21T23:17:20Z

Test build #90913 has finished for PR 21356 at commit fc8a197.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-05-21T23:19:04Z

Test build #90914 has finished for PR 21356 at commit 008d14d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2018-05-21T23:26:15Z

Merging to master / 2.3.

EventListeners can interrupt the event queue thread. In particular, when the EventLoggingListener writes to hdfs, hdfs can interrupt the thread. When there is an interrupt, the queue should be removed and stop accepting any more events. Before this change, the queue would continue to take more events (till it was full), and then would not stop when the application was complete because the PoisonPill couldn't be added. Added a unit test which failed before this change. Author: Imran Rashid <irashid@cloudera.com> Closes #21356 from squito/SPARK-24309. (cherry picked from commit 3244707) Signed-off-by: Marcelo Vanzin <vanzin@cloudera.com>

EventListeners can interrupt the event queue thread. In particular, when the EventLoggingListener writes to hdfs, hdfs can interrupt the thread. When there is an interrupt, the queue should be removed and stop accepting any more events. Before this change, the queue would continue to take more events (till it was full), and then would not stop when the application was complete because the PoisonPill couldn't be added. Added a unit test which failed before this change. Author: Imran Rashid <irashid@cloudera.com> Closes apache#21356 from squito/SPARK-24309. (cherry-picked from commit 3244707) Ref: LIHADOOP-40197 RB=1414281 BUG=LIHADOOP-40197 G=superfriends-reviewers R=fli,mshen,yezhou,edlu A=yezhou

vanzin reviewed May 17, 2018

View reviewed changes

review feedback

0a44c06

only remove the active listener for an interrupt

4a1f657

vanzin reviewed May 18, 2018

View reviewed changes

handle InterruptedException as well

09d55af

cloud-fan reviewed May 21, 2018

View reviewed changes

vanzin reviewed May 21, 2018

View reviewed changes

squito added 2 commits May 21, 2018 13:55

review feedback

fc8a197

grammar

008d14d

vanzin reviewed May 21, 2018

View reviewed changes

asfgit closed this in 3244707 May 21, 2018

squito deleted the SPARK-24309 branch May 23, 2018 17:47

wangshuo128 mentioned this pull request Nov 26, 2019

[SPARK-30059][CORE]Stop AsyncEventQueue when interrupted in dispatch #26674

Closed


		bus.start(mockSparkContext, mockMetricsSystem)

		// after we post one event, the shared queue should get stopped because of the interrupt

[SPARK-24309][CORE] AsyncEventQueue should stop on interrupt. #21356

[SPARK-24309][CORE] AsyncEventQueue should stop on interrupt. #21356

Conversation

squito commented May 17, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented May 18, 2018

cloud-fan commented May 18, 2018

cloud-fan commented May 18, 2018

squito commented May 18, 2018

SparkQA commented May 18, 2018

cloud-fan commented May 18, 2018

skonto commented May 18, 2018 • edited

squito commented May 18, 2018

squito commented May 18, 2018

skonto commented May 18, 2018 • edited

zsxwing commented May 18, 2018

vanzin commented May 18, 2018

zsxwing commented May 18, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vanzin commented May 18, 2018

zsxwing commented May 18, 2018

SparkQA commented May 18, 2018

squito commented May 18, 2018

vanzin commented May 18, 2018

SparkQA commented May 21, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vanzin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vanzin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented May 21, 2018

SparkQA commented May 21, 2018

vanzin commented May 21, 2018

skonto commented May 18, 2018 •

edited

skonto commented May 18, 2018 •

edited