Parallelise a for that uses views (Scala)

Question

I'm very new to Scala! However, I have the following working solution to Euler Problem 4 that I would like to use par on, just to see if I can do it:

import scala.math

object Problem4 {
    def isPalindrome(x: Int): Boolean = {
        val string = x.toString
        string.reverseIterator.sameElements(string.iterator)
    }

    def getPairs(minimum: Int, maximum: Int) = {
        for (i <- minimum to maximum view;
             j <- minimum to maximum view)
             yield (i, j)
    }

    def getAnswer(numberOfDigits: Int): Int = {
        val maximum = math.pow(10, numberOfDigits).toInt
        val minimum = math.pow(10, numberOfDigits - 1).toInt
        val products = for {
                           pair <- getPairs(minimum, maximum)
                           product = pair match { case (i, j) => i * j } 
                           if isPalindrome(product)
                       } yield product
        products.par.max
    }

    def main(args: Array[String]) {
        val answer = getAnswer(4)
        println("Problem 4 answer: %s".format(answer))
    }
} // object Problem4

Project Euler 4 asks for 3-digit numbers, and I've noticed that finding the answer for 4-digit numbers takes 63 seconds on my PC and only uses one processor on my dual-core system. This is despite applying par to the end of the for expression.

How do I parallelise this using par? Ideally I'd like finding the answer for 4 digits to take 30-40 seconds. Thanks!

EDIT: I'm pretty sure getPairs returns a View:

    scala>     def getPairs(minimum: Int, maximum: Int) = {
         |         for (i <- minimum to maximum view;
         |              j <- minimum to maximum view)
         |              yield (i, j)
         |     }
    getPairs: (minimum: Int, maximum: Int)scala.collection.SeqView[(Int, Int),Seq[_]]

Moreover, adding par to the getPairs call returns a warning, still only uses one of my processors, and results in a java.lang.OutOfMemoryError: Java heap space exception:

[info] Loading project definition from M:\programming\testdriveneuler\src\problem4\project
[info] Set current project to euler (in build file:/M:/programming/testdriveneuler/src/problem4/)
[info] Compiling 1 Scala source to M:\programming\testdriveneuler\src\problem4\target\scala-2.9.2\classes...
[warn] M:\programming\testdriveneuler\src\problem4\src\main\scala\Problem4.scala:39: `withFilter' method does not yet exist on scala.collection.parallel.ParSeq[((Int, Int), Int)], using `filter' method instead
[warn]                            pair <- getPairs(minimum, maximum).par
[warn]                                 ^
[warn] one warning found

EDIT: I'm explicitly interested in calculating the answer to Euler problem 4 for the product of 2 4-digit numbers. For reference the answer is 99000099.

Answer 1

So much complexity. It can be done with 2 functions

def isPalindrome(x: Int) = x.toString sameElements x.toString.reverse

def products(min: Int, max: Int) = for { 
  x <- min to max par; 
  y <- min to max par; 
  if isPalindrome(x * y) 
} yield (x, y, x * y)

scala> products(100, 999).maxBy(_._3)
res0: (Int, Int, Int) = (913,993,906609)

Update

(min to max).view returns SeqView, which represents lazy version of collection. (min to max).view.par returns ParSeq, parallel collection. In other words, calling par on lazy sequence will force it to evaluate. So, in this case you ought to choose between laziness and parallelism. It is hard to say what conversions are performed when you are moving from SeqView to ParSeq but this unnecessary complexity is causing OutOfMemoryError.

Update2

Yes, for is just a syntactic sugar over higher order operations on collections. Desugared version of for loop will be something like:

(min to max par) flatMap { x =>
  (min to max par)
    .filter(y => isPalindrome(x * y))
    .map(y => x * y)
}

Answer 2

Add the .par at the call to getPairs;

pair <- getPairs(min, max).par

I suspect (but can't be sure on my phone) that the getPairs method is not returning a view; hence your for-loop performs your computation.

A simple way of verifying this is to leave out the last line (i.e. products.par.max - the evaluation of the for-loop) and see whether your program still performs the computation.

Answer 3

You can parallelize it like this:

  def isPalindrome (n: Int) = n.toString == n.toString.reverse
  val R = 100 until 1000
  val xs = for (i <- R; j <- R) yield i * j
  val pals = xs.par filter isPalindrome
  println (pals max)

(leave out the .par for non-parallel). However I found the parallel version is 3-4 times slower on my dual-core machine. Sometimes the overhead of parallelizing isn't worth it.

Edit: for giggles, here's a version using Akka (based on the Pi calculation tutorial). Its performance is slightly faster than using parallel collections as in 4e6's answer (on my machine 8.0s vs 9.1s), although that solution becomes almost identical in performance if you remove the .par on the second generator.

import akka.actor._
import akka.routing.RoundRobinRouter

sealed trait Euler4Message
case object Calculate extends Euler4Message
case class Work(range1: Seq[Int], range2: Seq[Int]) extends Euler4Message
case class Result(value: Int) extends Euler4Message
case class FinalResult(value: Int, duration: Long)

class Worker extends Actor {

  def calculate(r1: Seq[Int], r2: Seq[Int]): Int = {
    def isPalindrome(x: Int) = {
      val s = x.toString
      s.reverseIterator.sameElements(s.iterator)
    }
    val pals = for (i <- r1; j <- r2; if isPalindrome(i * j)) yield i * j
    pals.max
  } 

  def receive = { case Work(r1, r2) => sender ! Result(calculate(r1, r2)) }   
}

class Master(nrOfDigits: Int, nrOfWorkers: Int, chunkSize: Int) extends Actor {

  var nrOfResults: Int = 0
  var maxResult = 0
  var sentAll = false
  var nrMessages = 0

  val start: Long = System.currentTimeMillis
  val min = math.pow(10, nrOfDigits - 1).toInt
  val max = math.pow(10, nrOfDigits).toInt 
  val range = min until max
  val workerRouter = 
    context.actorOf(Props[Worker].withRouter(RoundRobinRouter(nrOfWorkers)))

  def receive = {
    case Calculate =>
      for (i <- range.grouped(chunkSize)) { 
        // grouped produces an Iterator, so is 'lazy'
        workerRouter ! Work(i, range)
        nrMessages += 1
      }
      sentAll = true

    case Result(value) =>
      if (value > maxResult) maxResult = value
      nrOfResults += 1
      if (sentAll && nrOfResults == nrMessages) {
         println("Result = "+ maxResult 
                +"\nTime in ms: "+ (System.currentTimeMillis - start))
         context.system.shutdown()
      }
  }
}

object Euler4 extends App {
  val master =  ActorSystem().actorOf(Props(new Master(4, 4, 50)))
  master ! Calculate
}

The good thing about actors is that you can use imperative code and still get parallelism. So, swap out the calculate method in the worker actor above for the below, and the whole thing completes in about 1.0s (an 8x improvement).

I found it works fastest with larger chunk sizes (try 1000), and make sure you have at least as many workers as processors.

  def calculate(r1: Seq[Int], r2: Seq[Int]): Int = {
    def isPalindrome(x: Int) = {
      val s = x.toString
      s.reverseIterator.sameElements(s.iterator)
    }
    var max = 0
    // count down so that we don't have to check if palindrome so often
    var i = r1.last
    while (i >= r1.head) {
      // for some reason counting down here increases the run-time :/
      var j = r2.head
      while (j <= r2.last) {
        val r = i * j
        if (r > max && isPalindrome(r)) max = r
        j += 1
      }
      i -= 1
    }
    max
  }