raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Use nanoTime in FixedRateTask, see #2214

Browse source 
* Rewrote test to use latch and assert rate instead
This commit is contained in:
Patrik Nordwall 2012-06-12 13:37:21 +02:00
parent b27bae6554
commit 7b6ae2f5c9
2 changed files with 21 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

13
akka-cluster/src/main/scala/akka/cluster/FixedRateTask.scala
View file

@ -29,11 +29,10 @@ private[akka] object FixedRateTask {
*/ */
private[akka] class FixedRateTask(scheduler: Scheduler, initalDelay: Duration, delay: Duration, task: Runnable) extends Runnable { private[akka] class FixedRateTask(scheduler: Scheduler, initalDelay: Duration, delay: Duration, task: Runnable) extends Runnable {
private val delayMillis = delay.toMillis private val delayNanos = delay.toNanos
private val minDelayMillis = 1L
private val cancelled = new AtomicBoolean(false) private val cancelled = new AtomicBoolean(false)
private val counter = new AtomicLong(0L) private val counter = new AtomicLong(0L)
private val startTime = System.currentTimeMillis + initalDelay.toMillis private val startTime = System.nanoTime + initalDelay.toNanos
scheduler.scheduleOnce(initalDelay, this) scheduler.scheduleOnce(initalDelay, this)
def cancel(): Unit = cancelled.set(true) def cancel(): Unit = cancelled.set(true)
@ -41,11 +40,9 @@ private[akka] class FixedRateTask(scheduler: Scheduler, initalDelay: Duration, d
override final def run(): Unit = if (!cancelled.get) try { override final def run(): Unit = if (!cancelled.get) try {
task.run() task.run()
} finally if (!cancelled.get) { } finally if (!cancelled.get) {
val nextTime = startTime + delayMillis * counter.incrementAndGet val nextTime = startTime + delayNanos * counter.incrementAndGet
val nextDelayMillis = nextTime - System.currentTimeMillis // it's ok to schedule with negative duration, will run asap
val nextDelay = Duration( val nextDelay = Duration(nextTime - System.nanoTime, TimeUnit.NANOSECONDS)
(if (nextDelayMillis <= minDelayMillis) minDelayMillis else nextDelayMillis),
TimeUnit.MILLISECONDS)
try { try {
scheduler.scheduleOnce(nextDelay, this) scheduler.scheduleOnce(nextDelay, this)
} catch { case e: IllegalStateException ⇒ /* will happen when scheduler is closed, nothing wrong */ } } catch { case e: IllegalStateException ⇒ /* will happen when scheduler is closed, nothing wrong */ }

25
akka-cluster/src/test/scala/akka/cluster/FixedRateTaskSpec.scala
View file

@ -4,32 +4,39 @@
package akka.cluster package akka.cluster
import java.util.concurrent.atomic.AtomicInteger
import akka.testkit.AkkaSpec import akka.testkit.AkkaSpec
import akka.util.duration._ import akka.util.duration._
import akka.testkit.TimingTest import akka.testkit.TimingTest
import akka.testkit.TestLatch
import akka.dispatch.Await
@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner]) @org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class FixedRateTaskSpec extends AkkaSpec { class FixedRateTaskSpec extends AkkaSpec {
"Task scheduled at fixed rate" must { "Task scheduled at fixed rate" must {
"adjust for scheduler inaccuracy" taggedAs TimingTest in { "adjust for scheduler inaccuracy" taggedAs TimingTest in {
val counter = new AtomicInteger val startTime = System.nanoTime
val n = 33
val latch = new TestLatch(n)
FixedRateTask(system.scheduler, 150.millis, 150.millis) { FixedRateTask(system.scheduler, 150.millis, 150.millis) {
counter.incrementAndGet() latch.countDown()
} }
5000.millis.sleep() Await.ready(latch, 6.seconds)
counter.get must (be(33) or be(34)) val rate = n * 1000.0 / (System.nanoTime - startTime).nanos.toMillis
rate must be(6.66 plusOrMinus (0.4))
} }
"compensate for long running task" taggedAs TimingTest in { "compensate for long running task" taggedAs TimingTest in {
val counter = new AtomicInteger val startTime = System.nanoTime
val n = 22
val latch = new TestLatch(n)
FixedRateTask(system.scheduler, 225.millis, 225.millis) { FixedRateTask(system.scheduler, 225.millis, 225.millis) {
counter.incrementAndGet()
80.millis.sleep() 80.millis.sleep()
latch.countDown()
} }
5000.millis.sleep() Await.ready(latch, 6.seconds)
counter.get must (be(22) or be(23)) val rate = n * 1000.0 / (System.nanoTime - startTime).nanos.toMillis
rate must be(4.4 plusOrMinus (0.3))
} }
} }
} }