Akka/Scala example source code file (ConsistencySpec.scala)

This example Akka source code file (ConsistencySpec.scala) is included in my "Source Code Warehouse" project. The intent of this project is to help you more easily find Akka and Scala source code examples by using tags.

All credit for the original source code belongs to akka.io; I'm just trying to make examples easier to find. (For my Scala work, see my Scala examples and tutorials.)

Akka tags/keywords

actor, akka, cachemisaligned, concurrent, consistencycheckingactor, consistencyspec, dispatch, duration, int, last, long, test, testing, time, vars

The ConsistencySpec.scala Akka example source code

package akka.actor

import language.postfixOps

import akka.testkit.AkkaSpec
import akka.dispatch.{ ThreadPoolConfig, ThreadPoolConfigBuilder, UnboundedMailbox }
import scala.concurrent.duration._

object ConsistencySpec {
  val minThreads = 1
  val maxThreads = 2000
  val factor = 1.5d
  val threads = ThreadPoolConfig.scaledPoolSize(minThreads, factor, maxThreads) // Make sure we have more threads than cores

  val config = s"""
      consistency-dispatcher {
        throughput = 1
        executor = "fork-join-executor"
        fork-join-executor {
          parallelism-min = $minThreads
          parallelism-factor = $factor
          parallelism-max = $maxThreads
        }
      }
    """
  class CacheMisaligned(var value: Long, var padding1: Long, var padding2: Long, var padding3: Int) //Vars, no final fences

  class ConsistencyCheckingActor extends Actor {
    var left = new CacheMisaligned(42, 0, 0, 0) //var
    var right = new CacheMisaligned(0, 0, 0, 0) //var
    var lastStep = -1L
    def receive = {
      case step: Long ⇒

        if (lastStep != (step - 1))
          sender() ! "Test failed: Last step %s, this step %s".format(lastStep, step)

        var shouldBeFortyTwo = left.value + right.value
        if (shouldBeFortyTwo != 42)
          sender() ! "Test failed: 42 failed"
        else {
          left.value += 1
          right.value -= 1
        }

        lastStep = step
      case "done" ⇒ sender() ! "done"; context.stop(self)
    }
  }
}

class ConsistencySpec extends AkkaSpec(ConsistencySpec.config) {
  import ConsistencySpec._

  override def expectedTestDuration: FiniteDuration = 3.minutes

  "The Akka actor model implementation" must {
    "provide memory consistency" in {
      val noOfActors = threads + 1
      val props = Props[ConsistencyCheckingActor].withDispatcher("consistency-dispatcher")
      val actors = Vector.fill(noOfActors)(system.actorOf(props))

      for (i ← 0L until 100000L) {
        actors.foreach(_.tell(i, testActor))
      }

      for (a ← actors) { a.tell("done", testActor) }

      for (a ← actors) expectMsg(5 minutes, "done")
    }
  }
}