Solution to SPOJ PRIME1 using Scala Actors

SpojPrime1Actors.scala

import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.locks.ReentrantLock
import collection.immutable.HashMap
import Stream.cons
import java.io.{BufferedWriter, PrintWriter}
import Console.{readInt, readLine}
import Math.{sqrt, ceil}
import actors.Actor
import collection.mutable.ListBuffer

/*
 * My solution to SPOJ problem PRIME1 (https://www.spoj.pl/problems/PRIME1/) in Scala.
 */
object SpojPrime1Actors {

  def main(arguments: Array[String]) {
    val start = System.nanoTime
    val writer = new BufferedWriter(new PrintWriter(System.out))
    val noOfTestCases: Int = readInt
    val primesList = for {i<- 1 to noOfTestCases
      val input = readLine.trim.split(" ")
    } yield primesBetween(input(0) toInt, input(1) toInt)

    primesList foreach {primes=>
      primes() foreach { x: Int => writer.write("%s\n".format(x)) }
      writer.write("\n")
      writer.flush
    }
    
    writer.flush
    System.out.println("time taken = %s sec".format((System.nanoTime - start)/1e9))
  }

  /*
    Finds prime numbers between <code>start</code> and <code>end</code>.
   */
  def primesBetween(start: Int, end: Int): () => List[Int] = {
    val actor: PrimesBetweenActor = new PrimesBetweenActor(start, end)
    actor ! Start
    return { () => actor.getPrimesFound }
  }

  case class PrimeCheck(val number: Int, val checker: PrimesBetweenActor)
  case class PrimeResult(val number: Int, val result: Boolean)
  case object Start
  case object Stop

  class PrimesBetweenActor(val startNumber: Int, val endNumber: Int) extends Actor {
    private val noOfWorkers = 10
    private val workers = Stream
      .continually(1 to noOfWorkers map { i=> new IsPrimeActor(i) }).flatten

    private val primesFound: ListBuffer[Int] = ListBuffer()

    private val sentCount = new AtomicLong(0)
    private val receivedCount = new AtomicLong(0)
    private var sentAll = false
    private val lock = new ReentrantLock()
    private val done = lock.newCondition

    start
    def begin {
      val firstCandidate = startNumber match {
        case 1 => 3
        case 2 => 3
        case x => x match {
          case y if isPrime(y) => y
          case y if y % 2 == 0 => y + 1
          case y => y + 2
        }
      }

      lazy val candidates: Stream[Int] = cons(firstCandidate, candidates map { _ + 2 })
      lazy val realCandidates = if (startNumber == 1 || startNumber == 2) {
        cons(2, candidates)
      } else candidates

      realCandidates.takeWhile({ _ < endNumber }).zipWithIndex.foreach {z =>
        workers(z._2 % noOfWorkers) ! PrimeCheck(z._1, this)
        sentCount.incrementAndGet
      }
      sentAll = true
    }

    def act() {
      loop {
        react {
          case Start => begin
          case PrimeResult(n, r) => {
            if (r) primesFound.append(n)
            receivedCount.incrementAndGet
            lock.lock; done.signal; lock.unlock
          }
          case Stop => exit('stop)
        }
      }
    }

    def getPrimesFound = {
      lock.lock
      while (!(sentAll && sentCount.get == receivedCount.get)) {
        done.await
      }
      lock.unlock
      workers take noOfWorkers foreach { _ ! Stop }
      this ! Stop
      primesFound.sorted.toList
    }

  }

  class IsPrimeActor(val id: Int) extends Actor {
    start
    def act() {
      loop {
        react {
          case PrimeCheck(n, checker) => checker ! PrimeResult(n, isPrime(n))
          case Stop => exit('stop)
        }
      }
    }
  }

  private val odds: Stream[Int] = cons(3, odds map { _ + 2 })
  private val primes: Stream[Int] = cons(2, odds filter isPrime)

  val isPrime = memoize {
    (n: Int) => n match {
      case 1 => false
      case 2 => true
      case 3 => true
      case 5 => true
      case 7 => true
      case x => x match {
        case y if y % 3 == 0 => false
        case y if y % 5 == 0 => false
        case y if y % 7 == 0 => false
        case y if !((y + 1) % 6 == 0 || (y - 1) % 6 == 0) => false
        case y => primeDivisors(y) isEmpty
      }
    }
  }

  private def primeDivisors(n: Int) =
    primes takeWhile { _ <= ceil(sqrt(n))} filter {n % _ == 0 }

  def memoize[X,R](f: X => R) = {
    var cache = new HashMap[X,R];
    { (x: X) => {
      if (cache contains x) cache.get(x).get
      else synchronized {
        if (cache contains x) cache.get(x).get
        else {
          val fx = f(x)
          cache = cache + (x -> fx)
          fx
        }
      }
    }}
  }

}