[SPARK-4813][Streaming] Fix the issue that ContextWaiter didn't handle 'spurious wakeup' #3661

zsxwing · 2014-12-10T08:49:16Z

Used Condition to rewrite ContextWaiter because it provides a convenient API awaitNanos for timeout.

SparkQA · 2014-12-10T08:55:24Z

Test build #24302 has started for PR 3661 at commit e06bd4f.

This patch merges cleanly.

srowen · 2014-12-10T09:25:37Z

streaming/src/main/scala/org/apache/spark/streaming/ContextWaiter.scala

+  private val lock = new ReentrantLock()
+  private val condition = lock.newCondition()
+
+  @GuardedBy("lock")


Minor point - these are not in the JDK but in a Findbugs library for JSR-305. It's not used in Spark, and happens to be a dependency now. Maybe not worth using it just 1 place?

Maybe not worth using it just 1 place?

So which one do you prefer?

Use comments to describe such information.

Use GuardedBy from now on.

In addition, now Findbugs does not recognize GuardedBy in Scala codes.

BTW, I turned to GuardedBy because @aarondav asked me to do it in #3634

Yes, that's why I brought it up. It's not actually a standard Java annotation (unless someone tells me it just turned up in 8 or something) but part of JSR-305. This is a dependency of Spark core at the moment, but none of the annotations are used. I think we should just not use them instead of using this lib in 1 place.

OK. I changed them to comments.

In general, I'm a fan of the FindBugs annotations. I've had trouble getting the various analysis tools to work well with them in Scala, though. +1 to this commenting convention, though; this is very helpful.

SparkQA · 2014-12-10T10:05:25Z

Test build #24305 has started for PR 3661 at commit be42bcf.

This patch merges cleanly.

SparkQA · 2014-12-10T10:14:45Z

Test build #24302 has finished for PR 3661 at commit e06bd4f.

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

AmplabJenkins · 2014-12-10T10:14:48Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24302/
Test PASSed.

SparkQA · 2014-12-10T11:23:12Z

Test build #24305 has finished for PR 3661 at commit be42bcf.

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

AmplabJenkins · 2014-12-10T11:23:16Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24305/
Test PASSed.

JoshRosen · 2014-12-16T20:57:57Z

streaming/src/main/scala/org/apache/spark/streaming/ContextWaiter.scala

-  def notifyStop() = synchronized {
-    stopped = true
-    notifyAll()
+  def notifyError(e: Throwable) = {


Minor nit, but mind adding : Unit here?

JoshRosen · 2014-12-16T21:02:06Z

Maybe a naive question, but is there a reason why we can't just use synchronized methods here?

zsxwing · 2014-12-17T01:26:39Z

Maybe a naive question, but is there a reason why we can't just use synchronized methods here?

I wrote the reason in the description: "Used Condition to rewrite ContextWaiter because it provides a convenient API awaitNanos for timeout."

SparkQA · 2014-12-17T01:37:25Z

Test build #24521 has started for PR 3661 at commit 52247f5.

This patch merges cleanly.

SparkQA · 2014-12-17T02:58:40Z

Test build #24521 has finished for PR 3661 at commit 52247f5.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class Analyzer(catalog: Catalog, registry: FunctionRegistry, caseSensitive: Boolean)

AmplabJenkins · 2014-12-17T02:58:43Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24521/
Test PASSed.

JoshRosen · 2014-12-17T21:46:42Z

streaming/src/main/scala/org/apache/spark/streaming/ContextWaiter.scala

+   * Return `true` if it's stopped; or throw the reported error if `notifyError` has been called; or
+   * `false` if the waiting time detectably elapsed before return from the method.
+   */
+  def waitForStopOrError(timeout: Long = -1): Boolean = {


I considered whether this should throw TimeoutException instead of signaling failure via a boolean, but I guess this is sort of like Object.wait() or Condition.wait(), both of which don't throw exceptions.

Also, it looks like the usages of this in StreamingContext support this theory:

/** * Wait for the execution to stop. Any exceptions that occurs during the execution * will be thrown in this thread. */ def awaitTermination() { waiter.waitForStopOrError() } /** * Wait for the execution to stop. Any exceptions that occurs during the execution * will be thrown in this thread. * @param timeout time to wait in milliseconds */ def awaitTermination(timeout: Long) { waiter.waitForStopOrError(timeout) }

I suppose the documentation for the SparkContext.awaitTermination(timeout) could be improved to convey what happens when the timeout occurs...

Do you think if it's better to return Boolean to indicate if it's timeout? Although it will break the source compatibility.

Another option would be to just add a new def isTerminated: Boolean method, which would let users write code like

waiter.awaitTermination(1000) if (!waiter.isTerminated) { throw Exception(...) }

Agreed. Done.

How about making awaitTermination throw a TimeoutException if timeout? It looks a better API.

@tdas what do you think?

I think that will break a lot of existing programs for people. A lot of time (even in our own test cases) awaitTermination(timeout) is used for wait for a short period of time before check checking status or something. Currently that times out return quietly. If instead it starts throwing exceptions, then it will completely break some applications. So I strongly advise against changing this behavior.

In hindsight, instead of modeling awaitTermination against Akka ActorSystem's awaitTermination (which return Unit) , I should have modeled it like Java ExecutorService's awaitTermination which returns a Boolean. Now its not possible to change the API without breaking compatiblity. :(

In hindsight, instead of modeling awaitTermination against Akka ActorSystem's awaitTermination (which return Unit) , I should have modeled it like Java ExecutorService's awaitTermination which returns a Boolean. Now its not possible to change the API without breaking compatiblity. :(

@tdas, Sorry that I forgot to reply you. You said you designed it just like Akka ActorSystem.awaitTermination. But ActorSystem.awaitTermination will throw a TimeoutException in case of timeout.

/** * Block current thread until the system has been shutdown, or the specified * timeout has elapsed. This will block until after all on termination * callbacks have been run. * * @throws TimeoutException in case of timeout */ def awaitTermination(timeout: Duration): Unit

JoshRosen · 2014-12-17T21:49:15Z

This looks good to me.

SparkQA · 2014-12-18T03:27:27Z

Test build #24568 has started for PR 3661 at commit 15357d2.

This patch merges cleanly.

SparkQA · 2014-12-18T04:51:38Z

Test build #24568 has finished for PR 3661 at commit 15357d2.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class Analyzer(catalog: Catalog, registry: FunctionRegistry, caseSensitive: Boolean)

AmplabJenkins · 2014-12-18T04:51:42Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24568/
Test PASSed.

tdas · 2014-12-29T21:19:16Z

This almost looks good. Can you add a unit test in the StreamingContextSuite to test the new public API isTerminated (with and without error)? You could update the existing unit tests.

tdas · 2014-12-29T21:27:44Z

Actually, correction. Can you put the public API isTerminated on a different PR? I want to separate the addition to public API (which requires a design discussion) from this bug fix PR. In that case, there is no need to update unit tests, and this PR will be good to go.

SparkQA · 2014-12-30T02:02:36Z

Test build #24876 has started for PR 3661 at commit 52247f5.

This patch merges cleanly.

zsxwing · 2014-12-30T02:03:39Z

I deleted the commit about isTerminated. It will be sent in a different PR.

SparkQA · 2014-12-30T03:06:32Z

Test build #24876 has finished for PR 3661 at commit 52247f5.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-12-30T03:06:36Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24876/
Test FAILed.

zsxwing · 2014-12-30T04:15:24Z

Jenkins, retest this please.

SparkQA · 2014-12-30T04:17:30Z

Test build #24881 has started for PR 3661 at commit 52247f5.

This patch merges cleanly.

SparkQA · 2014-12-30T05:43:36Z

Test build #24881 has finished for PR 3661 at commit 52247f5.

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

AmplabJenkins · 2014-12-30T05:43:39Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24881/
Test PASSed.

tdas · 2014-12-30T22:38:40Z

Merging this. Thanks!

…e 'spurious wakeup' Used `Condition` to rewrite `ContextWaiter` because it provides a convenient API `awaitNanos` for timeout. Author: zsxwing <zsxwing@gmail.com> Closes #3661 from zsxwing/SPARK-4813 and squashes the following commits: 52247f5 [zsxwing] Add explicit unit type be42bcf [zsxwing] Update as per review suggestion e06bd4f [zsxwing] Fix the issue that ContextWaiter didn't handle 'spurious wakeup' (cherry picked from commit 6a89782) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

Fix the issue that ContextWaiter didn't handle 'spurious wakeup'

e06bd4f

srowen reviewed Dec 10, 2014
View reviewed changes

Update as per review suggestion

be42bcf

JoshRosen reviewed Dec 16, 2014
View reviewed changes

Add explicit unit type

52247f5

JoshRosen reviewed Dec 17, 2014
View reviewed changes

asfgit closed this in 6a89782 Dec 30, 2014

zsxwing deleted the SPARK-4813 branch December 31, 2014 01:31

[SPARK-4813][Streaming] Fix the issue that ContextWaiter didn't handle 'spurious wakeup' #3661

[SPARK-4813][Streaming] Fix the issue that ContextWaiter didn't handle 'spurious wakeup' #3661

Conversation

zsxwing commented Dec 10, 2014

SparkQA commented Dec 10, 2014

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

Choose a reason for hiding this comment

SparkQA commented Dec 10, 2014

SparkQA commented Dec 10, 2014

AmplabJenkins commented Dec 10, 2014

SparkQA commented Dec 10, 2014

AmplabJenkins commented Dec 10, 2014

Choose a reason for hiding this comment

JoshRosen commented Dec 16, 2014

zsxwing commented Dec 17, 2014

SparkQA commented Dec 17, 2014

SparkQA commented Dec 17, 2014

AmplabJenkins commented Dec 17, 2014

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

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

JoshRosen commented Dec 17, 2014

SparkQA commented Dec 18, 2014

SparkQA commented Dec 18, 2014

AmplabJenkins commented Dec 18, 2014

tdas commented Dec 29, 2014

tdas commented Dec 29, 2014

SparkQA commented Dec 30, 2014

zsxwing commented Dec 30, 2014

SparkQA commented Dec 30, 2014

AmplabJenkins commented Dec 30, 2014

zsxwing commented Dec 30, 2014

SparkQA commented Dec 30, 2014

SparkQA commented Dec 30, 2014

AmplabJenkins commented Dec 30, 2014

tdas commented Dec 30, 2014