[SPARK-10224][Streaming]Fix the issue that blockIntervalTimer won't call updateCurrentBuffer when stopping #8417

zsxwing · 2015-08-25T13:34:56Z

blockIntervalTimer.stop(interruptTimer = false) doesn't guarantee calling updateCurrentBuffer. So it's possible that blockIntervalTimer will exit when updateCurrentBuffer is not empty. Then the data in currentBuffer will be lost.

To reproduce it, you can add Thread.sleep(200) in this line (

spark/streaming/src/main/scala/org/apache/spark/streaming/util/RecurringTimer.scala

Line 100 in 69c9c17

logDebug("Callback for " + name + " called at time " + prevTime)

) and run StreamingContexSuite.
I cannot write a unit test to reproduce it because I cannot find an approach to force RecurringTimer suspend at this line for a few milliseconds.

There was a failure in Jenkins here: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/41455/console

This PR updates RecurringTimer to make sure stop(interruptTimer = false) will call callback at least once after the stop method is called.

…when stopping

zsxwing · 2015-08-25T13:37:22Z

streaming/src/test/scala/org/apache/spark/streaming/receiver/BlockGeneratorSuite.scala

@@ -184,9 +184,10 @@ class BlockGeneratorSuite extends SparkFunSuite with BeforeAndAfter {
    // Verify that the final data is present in the final generated block and
    // pushed before complete stop
    assert(blockGenerator.isStopped() === false) // generator has not stopped yet
-    clock.advance(blockIntervalMs)   // force block generation


There is a race condition here. If updateCurrentBuffer finishes before blockIntervalTimer.stop(interruptTimer = false), this test will fail because RecurringTimer is blocking in this line:

spark/streaming/src/main/scala/org/apache/spark/streaming/util/RecurringTimer.scala

Line 96 in 69c9c17

clock.waitTillTime(nextTime)

Therefore I called clock.advance(blockIntervalMs) in every trial to make sure we wont' block in clock.waitTillTime.

SparkQA · 2015-08-25T14:31:06Z

Test build #41534 has finished for PR 8417 at commit 94f108b.

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

andrewor14 · 2015-09-01T20:30:31Z

@tdas

zsxwing · 2015-09-21T15:06:13Z

@tdas let me try to use the following table to explain the issue. Assume currentBuffer is empty at the beginning, and there are 3 threads running in the following execution order:

execution order	thread 1 (RecurringTimer.loop())	thread 2 (BlockGenerator.stop())	thread 3
1	prevTime = nextTime
2	nextTime += period
3	logDebug("Callback for " + name + " called at time " + prevTime)
4			BlockGenerator.addData("1")
5			BlockGenerator.addData("2")
6		state = StoppedAddingData
7		blockIntervalTimer.stop(interruptTimer = false)
8	while (!stopped) {
9	exit RecurringTimer.loop

After step 9, currentBuffer is [1, 2] but won't be consumed since RecurringTimer has already exited.

harishreedharan · 2015-09-21T19:01:13Z

streaming/src/main/scala/org/apache/spark/streaming/receiver/BlockGenerator.scala

+    var isCurrentBufferEmpty = synchronized { currentBuffer.isEmpty }
+    while(!isCurrentBufferEmpty) {
+      Thread.sleep(blockIntervalMs)
+      isCurrentBufferEmpty = synchronized { currentBuffer.isEmpty }


A better fix would be to ensure that updateCurrentBuffer is called when blockIntervalTimer is called. This can be done by simply adding a parameter to RecurringTimer to ensure that the callback is made in the stop method.

The advantage is that we won't be burning CPU here.

tdas · 2015-09-21T21:28:47Z

@zsxwing I got the race condition already, but thank for adding that table, makes it very very obvious. Regarding solution, I think I agree with @harishreedharan at a high level. It is better to ensure that the existing assumption on RecurringTimer is better met. There are number of uses of RecurringTimer and its best if that gives more guarantees, rather than higher layers work around its limitations. So could you make sure the it is guaranteed that RecurringTimer will call the function one more time after stop is called?

zsxwing · 2015-09-22T01:32:26Z

How about adding an onStop parameter like this:

class RecurringTimer(clock: Clock, period: Long, callback: (Long) => Unit, onStop: (Long) => Unit, name: String)

So that we could do different work when stopping RecurringTimer. onStop is just doing nothing by default.

tdas · 2015-09-22T01:37:36Z

Why add another function? Cant we just ensure that the callback gets called once more after timer.stop(immediately = false)?

zsxwing · 2015-09-22T01:51:56Z

Why add another function? Cant we just ensure that the callback gets called once more after timer.stop(immediately = false)?

The semantics of RecurringTimer is calling callback every period milliseconds. It looks weird that we call callback after stopping.

tdas · 2015-09-22T02:35:02Z

Oh yes, it will wait for the right amount of time before calling the "callback" one last time.

SparkQA · 2015-09-22T03:47:50Z

Test build #42807 has finished for PR 8417 at commit a1c9354.

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

harishreedharan · 2015-09-22T04:27:22Z

streaming/src/main/scala/org/apache/spark/streaming/util/RecurringTimer.scala

@@ -99,6 +99,8 @@ class RecurringTimer(clock: Clock, period: Long, callback: (Long) => Unit, name:
        nextTime += period
        logDebug("Callback for " + name + " called at time " + prevTime)
      }
+      clock.waitTillTime(nextTime)


Do we really need to wait till the next time? You could just take a parameter specifying whether the method should be called on stop or not, and then pass the current time in.

Do we really need to wait till the next time? You could just take a parameter specifying whether the method should be called on stop or not, and then pass the current time in.

I just want to make it not break the semantics of RecurringTimer.

Yes, the RecurringTimer is designed to call the function at particular time interval, it should not deviate from that even for the last call.

BTW, are you updating the prevTime which is returned by stop() as the last time the call was made?
I think there arent unit tests for this, could you add them, so that this behavior is properly captured.

zsxwing · 2015-09-22T04:29:34Z

@tdas updated this PR. Could you take another look?

SparkQA · 2015-09-22T15:33:53Z

Test build #42837 has finished for PR 8417 at commit 193d191.

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

SparkQA · 2015-09-22T15:45:28Z

Test build #42838 has finished for PR 8417 at commit 429728f.

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

tdas · 2015-09-22T21:29:17Z

streaming/src/test/scala/org/apache/spark/streaming/util/RecurringTimerSuite.scala

+    }
+    @volatile var lastTime = -1L
+    // Now RecurringTimer is waiting for the next interval
+    val t = new Thread {


super nit: t -> thread, got confused with t being timer.

zsxwing · 2015-09-23T00:43:53Z

I removed isStopped since it's only for testing. I used PrivateMethod in unit tests to access stopped rather than adding isStopped in this PR.

SparkQA · 2015-09-23T01:33:24Z

Test build #42874 has finished for PR 8417 at commit e8e490d.

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

tdas · 2015-09-23T01:54:44Z

LGTM. I am running the tests a few times to make sure that this its not flaky.

tdas · 2015-09-23T02:23:47Z

Could you update the solution in the PR text to specify the actual solution.

SparkQA · 2015-09-23T02:32:07Z

Test build #1787 has finished for PR 8417 at commit e8e490d.

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

SparkQA · 2015-09-23T02:45:02Z

Test build #1786 has finished for PR 8417 at commit e8e490d.

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

SparkQA · 2015-09-23T03:04:22Z

Test build #1788 has finished for PR 8417 at commit e8e490d.

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

SparkQA · 2015-09-23T03:17:19Z

Test build #1789 has finished for PR 8417 at commit e8e490d.

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

zsxwing · 2015-09-23T05:45:41Z

Test build #1787 has finished for PR 8417 at commit e8e490d.

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

This is not related to this patch. I fixed it in #8877.

SparkQA · 2015-09-23T06:06:21Z

Test build #1792 has finished for PR 8417 at commit e8e490d.

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

SparkQA · 2015-09-23T06:38:41Z

Test build #1790 has finished for PR 8417 at commit e8e490d.

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

tdas · 2015-09-23T06:41:06Z

I am merging this to master and 1.5.

SparkQA · 2015-09-23T06:42:32Z

Test build #1791 has finished for PR 8417 at commit e8e490d.

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

… call updateCurrentBuffer when stopping `blockIntervalTimer.stop(interruptTimer = false)` doesn't guarantee calling `updateCurrentBuffer`. So it's possible that `blockIntervalTimer` will exit when `updateCurrentBuffer` is not empty. Then the data in `currentBuffer` will be lost. To reproduce it, you can add `Thread.sleep(200)` in this line (https://github.com/apache/spark/blob/69c9c177160e32a2fbc9b36ecc52156077fca6fc/streaming/src/main/scala/org/apache/spark/streaming/util/RecurringTimer.scala#L100) and run `StreamingContexSuite`. I cannot write a unit test to reproduce it because I cannot find an approach to force `RecurringTimer` suspend at this line for a few milliseconds. There was a failure in Jenkins here: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/41455/console This PR updates RecurringTimer to make sure `stop(interruptTimer = false)` will call `callback` at least once after the `stop` method is called. Author: zsxwing <zsxwing@gmail.com> Closes #8417 from zsxwing/SPARK-10224. (cherry picked from commit 44c28ab) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

… call updateCurrentBuffer when stopping `blockIntervalTimer.stop(interruptTimer = false)` doesn't guarantee calling `updateCurrentBuffer`. So it's possible that `blockIntervalTimer` will exit when `updateCurrentBuffer` is not empty. Then the data in `currentBuffer` will be lost. To reproduce it, you can add `Thread.sleep(200)` in this line (https://github.com/apache/spark/blob/69c9c177160e32a2fbc9b36ecc52156077fca6fc/streaming/src/main/scala/org/apache/spark/streaming/util/RecurringTimer.scala#L100) and run `StreamingContexSuite`. I cannot write a unit test to reproduce it because I cannot find an approach to force `RecurringTimer` suspend at this line for a few milliseconds. There was a failure in Jenkins here: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/41455/console This PR updates RecurringTimer to make sure `stop(interruptTimer = false)` will call `callback` at least once after the `stop` method is called. Author: zsxwing <zsxwing@gmail.com> Closes apache#8417 from zsxwing/SPARK-10224. (cherry picked from commit 44c28ab) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com> (cherry picked from commit 6a616d0)

Fix the issue that blockIntervalTimer won't call updateCurrentBuffer …

94f108b

…when stopping

zsxwing reviewed Aug 25, 2015
View reviewed changes

harishreedharan reviewed Sep 21, 2015
View reviewed changes

Make RecurringTimer call callback again after stopping

a1c9354

harishreedharan reviewed Sep 22, 2015
View reviewed changes

zsxwing added 2 commits September 22, 2015 22:36

Fix RecurringTimer and add unit tests

193d191

Fix a typo

429728f

tdas reviewed Sep 22, 2015
View reviewed changes

Remove isStopped and use PrivateMethod to access it

e8e490d

asfgit closed this in 44c28ab Sep 23, 2015

zsxwing deleted the SPARK-10224 branch September 23, 2015 12:43

[SPARK-10224][Streaming]Fix the issue that blockIntervalTimer won't call updateCurrentBuffer when stopping #8417

[SPARK-10224][Streaming]Fix the issue that blockIntervalTimer won't call updateCurrentBuffer when stopping #8417

Uh oh!

Conversation

zsxwing commented Aug 25, 2015

Uh oh!

zsxwing Aug 25, 2015

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 25, 2015

Uh oh!

andrewor14 commented Sep 1, 2015

Uh oh!

zsxwing commented Sep 21, 2015

Uh oh!

harishreedharan Sep 21, 2015

Choose a reason for hiding this comment

Uh oh!

tdas commented Sep 21, 2015

Uh oh!

zsxwing commented Sep 22, 2015

Uh oh!

tdas commented Sep 22, 2015

Uh oh!

zsxwing commented Sep 22, 2015

Uh oh!

tdas commented Sep 22, 2015

Uh oh!

SparkQA commented Sep 22, 2015

Uh oh!

harishreedharan Sep 22, 2015

Choose a reason for hiding this comment

Uh oh!

zsxwing Sep 22, 2015

Choose a reason for hiding this comment

Uh oh!

tdas Sep 22, 2015

Choose a reason for hiding this comment

Uh oh!

tdas Sep 22, 2015

Choose a reason for hiding this comment

Uh oh!

zsxwing commented Sep 22, 2015

Uh oh!

SparkQA commented Sep 22, 2015

Uh oh!

SparkQA commented Sep 22, 2015

Uh oh!

tdas Sep 22, 2015

Choose a reason for hiding this comment

Uh oh!

zsxwing commented Sep 23, 2015

Uh oh!

SparkQA commented Sep 23, 2015

Uh oh!

tdas commented Sep 23, 2015

Uh oh!

tdas commented Sep 23, 2015

Uh oh!

SparkQA commented Sep 23, 2015

Uh oh!

SparkQA commented Sep 23, 2015

Uh oh!

SparkQA commented Sep 23, 2015

Uh oh!

SparkQA commented Sep 23, 2015

Uh oh!

zsxwing commented Sep 23, 2015

Uh oh!

SparkQA commented Sep 23, 2015

Uh oh!

SparkQA commented Sep 23, 2015

Uh oh!

tdas commented Sep 23, 2015

Uh oh!

SparkQA commented Sep 23, 2015

Uh oh!

Uh oh!