Saturday, October 24, 2009

Dual core acting in Scala

Having a dual core processor is a great thing for the operating system; but how can you make use of this cool technology as a (mere mortal) programmer? Potentially a complex issue but if you use Scala you'll be pleasantly surprised. It is the concept of Actors and message passing that saves the day. Let me explain using an example from mathematics.

A problem that can theoretically benefit from multiple cores is the one that calculates the integral of a function. Using the simple quadrature method this problem is reduced to calculating the area of all the rectangles that follow the curve, and adding these up to produce an estimate of the integral. More rectangles provide a more accurate evaluation. The image (from wikipedia) should help you get the picture.
Here is that algorithm coded in Scala: 

type R = Double
def integrate(f : R=>R , a : R, b : R, intervalCount : Int ) = {  
val intervals = (1 to intervalCount)
val width = (b-a) / intervalCount  
val halfW = width/2.0  
(0.0 /: intervals) {
  (s,i) => s + width * f(a + i * width - halfW)
}}





Note that I created a type alias for Double called R. An interesting piece of code here is the last line of the function. The /: operator is the foldleft operator. This foldleft starts with a value of 0.0 and calculates the sum of width * f(a + i * width - halfW) for each i in intervals.


To calculate the integral of the Sine function from 0 to 90 degrees using 100 rectangles , the Scala function call looks like this: 


integrate(Math.sin,0,Math.toRadians(90),100)


Note that I pass the Java sine function as an argument. The answer is 1.0000102809119051 Increasing the number of rectangles to 10000 gives a better estimate: 1.0000000010280816.


Here's how I can use my dual core CPU to calculate the integral faster: divide the calculation into two smaller calculations and get each core to calculate the result of a smaller part; then add up the two results and voila I am done. The two parts are defined by using the midpoint between a and b and allowing one core work on the rectangles less than this midpoint while the other core chows the rectangles that are greater than the midpoint . Thus, if I want 10 000 rectangles, each core would do 5000 calculations.


Here's the divide and conquer algorithm in Scala: 



val m = (b - a) / 2.0
val caller = scala.actors.Actor.self
val iH = intervals / 2
scala.actors.Actor.actor{caller!integrate(f,a,m,iH)}
scala.actors.Actor.actor{caller!integrate(f,m,b,iH)}
val p1 = scala.actors.Actor.receive {case a : R => a}
val p2 = scala.actors.Actor.receive {case a : R => a}
p1 + p2



The first thing to note is that there is an implicit actor called self. The ! operator combined with the actor method sends a message to the actors queue and processes it asynchronously. When these processes complete they send the result back to the caller. The caller (i.e. self) receives these messages and assigned the values p1 and p2 to the messages.

The full program is given below. I optimised the final code a bit to use a tail recursive function instead of a range.


I am sure you get the picture. Scala actors are real stars! 



package scalablogs





object Application {

  type R = Double

  val intervals = 10000000



  def integrate(f : R=>R ,  a : R,  b : R, intervalCount : Int )  = {

    val width = (b-a) / intervalCount

    val halfW = width/2.0

    def calc(i : Int, t : R) : R = {

      val area = width * f(a + (i+1) * width - halfW)

      if ( i < 2)

         t + area

      else

         calc(i-1,t+area)

    }

    val r = calc(intervalCount-1,0)

    r

  }



  def integrateNormal(f : R=>R ,  a : R,  b : R)  = {

    integrate(f,a,b,intervals)

  }



  def integrateConcurrent(f : R=>R ,  a : R,  b : R ) = {

    val m = (b - a) / 2.0

    val caller = scala.actors.Actor.self

    val iH = intervals / 2

    scala.actors.Actor.actor{caller!integrate(f,a,m,iH)}
   scala.actors.Actor.actor{caller!integrate(f,m,b,iH)}

    val p1 = scala.actors.Actor.receive {case a : R => a}

    val p2 = scala.actors.Actor.receive {case a : R => a}

    p1 + p2

  }



  def main(args : Array[String]) = {

    val start = System.nanoTime()

    val r = integrateNormal(Math.sin,0,Math.toRadians(90))

    val elapsed = (System.nanoTime - start)/1000000000.0

    println("Normal answer is " + r + " (" + elapsed + " second(s))")

    val start2 = System.nanoTime()

    val r2 = integrateConcurrent(Math.sin,0,Math.toRadians(90))

    val elapsed2 = (System.nanoTime - start2)/1000000000.0

    println("Concurrent answer is " + r2+ " (" + elapsed2 + " second(s))")

  }



}

Posted by at 10:58 AM
Labels:

0 comments:

Post a Comment

Subscribe to: Post Comments (Atom)