You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by jo...@apache.org on 2015/04/03 07:54:50 UTC
spark git commit: [SPARK-6621][Core] Fix the bug that calling
EventLoop.stop in EventLoop.onReceive/onError/onStart doesn't call onStop
Repository: spark
Updated Branches:
refs/heads/master 6e1c1ec67 -> 440ea31b7
[SPARK-6621][Core] Fix the bug that calling EventLoop.stop in EventLoop.onReceive/onError/onStart doesn't call onStop
Author: zsxwing <zs...@gmail.com>
Closes #5280 from zsxwing/SPARK-6621 and squashes the following commits:
521125e [zsxwing] Fix the bug that calling EventLoop.stop in EventLoop.onReceive and EventLoop.onError doesn't call onStop
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/440ea31b
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/440ea31b
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/440ea31b
Branch: refs/heads/master
Commit: 440ea31b76aa7e813436271fd63880c7bcd69157
Parents: 6e1c1ec
Author: zsxwing <zs...@gmail.com>
Authored: Thu Apr 2 22:54:30 2015 -0700
Committer: Josh Rosen <jo...@databricks.com>
Committed: Thu Apr 2 22:54:30 2015 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/util/EventLoop.scala | 18 ++++-
.../org/apache/spark/util/EventLoopSuite.scala | 72 ++++++++++++++++++++
2 files changed, 87 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/440ea31b/core/src/main/scala/org/apache/spark/util/EventLoop.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/EventLoop.scala b/core/src/main/scala/org/apache/spark/util/EventLoop.scala
index b0ed908..e9b2b8d 100644
--- a/core/src/main/scala/org/apache/spark/util/EventLoop.scala
+++ b/core/src/main/scala/org/apache/spark/util/EventLoop.scala
@@ -76,9 +76,21 @@ private[spark] abstract class EventLoop[E](name: String) extends Logging {
def stop(): Unit = {
if (stopped.compareAndSet(false, true)) {
eventThread.interrupt()
- eventThread.join()
- // Call onStop after the event thread exits to make sure onReceive happens before onStop
- onStop()
+ var onStopCalled = false
+ try {
+ eventThread.join()
+ // Call onStop after the event thread exits to make sure onReceive happens before onStop
+ onStopCalled = true
+ onStop()
+ } catch {
+ case ie: InterruptedException =>
+ Thread.currentThread().interrupt()
+ if (!onStopCalled) {
+ // ie is thrown from `eventThread.join()`. Otherwise, we should not call `onStop` since
+ // it's already called.
+ onStop()
+ }
+ }
} else {
// Keep quiet to allow calling `stop` multiple times.
}
http://git-wip-us.apache.org/repos/asf/spark/blob/440ea31b/core/src/test/scala/org/apache/spark/util/EventLoopSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/util/EventLoopSuite.scala b/core/src/test/scala/org/apache/spark/util/EventLoopSuite.scala
index 1026cb2..47b5352 100644
--- a/core/src/test/scala/org/apache/spark/util/EventLoopSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/EventLoopSuite.scala
@@ -203,4 +203,76 @@ class EventLoopSuite extends FunSuite with Timeouts {
assert(!eventLoop.isActive)
}
}
+
+ test("EventLoop: stop() in onStart should call onStop") {
+ @volatile var onStopCalled: Boolean = false
+ val eventLoop = new EventLoop[Int]("test") {
+
+ override def onStart(): Unit = {
+ stop()
+ }
+
+ override def onReceive(event: Int): Unit = {
+ }
+
+ override def onError(e: Throwable): Unit = {
+ }
+
+ override def onStop(): Unit = {
+ onStopCalled = true
+ }
+ }
+ eventLoop.start()
+ eventually(timeout(5 seconds), interval(5 millis)) {
+ assert(!eventLoop.isActive)
+ }
+ assert(onStopCalled)
+ }
+
+ test("EventLoop: stop() in onReceive should call onStop") {
+ @volatile var onStopCalled: Boolean = false
+ val eventLoop = new EventLoop[Int]("test") {
+
+ override def onReceive(event: Int): Unit = {
+ stop()
+ }
+
+ override def onError(e: Throwable): Unit = {
+ }
+
+ override def onStop(): Unit = {
+ onStopCalled = true
+ }
+ }
+ eventLoop.start()
+ eventLoop.post(1)
+ eventually(timeout(5 seconds), interval(5 millis)) {
+ assert(!eventLoop.isActive)
+ }
+ assert(onStopCalled)
+ }
+
+ test("EventLoop: stop() in onError should call onStop") {
+ @volatile var onStopCalled: Boolean = false
+ val eventLoop = new EventLoop[Int]("test") {
+
+ override def onReceive(event: Int): Unit = {
+ throw new RuntimeException("Oops")
+ }
+
+ override def onError(e: Throwable): Unit = {
+ stop()
+ }
+
+ override def onStop(): Unit = {
+ onStopCalled = true
+ }
+ }
+ eventLoop.start()
+ eventLoop.post(1)
+ eventually(timeout(5 seconds), interval(5 millis)) {
+ assert(!eventLoop.isActive)
+ }
+ assert(onStopCalled)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org