Divide and Accumulate with Scala Actors

In Scala we can easily do things in parallel with actors; actors are easily started up, they are cheap, and they can easily send objects to each other
.
Here I'll look at how to use actors to parallelize in situations where we need to do some sort of calculation and aggregation over a collection of objects. E.g. given a collection of RSS or ATOM feeds pull down the list recents items from each one and build the aggregated list of items from all the feeds ordered by publication time. Doing that sequentially is sort of trivial, and involves waiting for a request to each feed one at a time. It would by nice to fire all those requests in parallel, then wait for them to return thier results and build the aggregated list as the results arrive. The following diagram shows how to do that with actors:

Let the program start an actor per feed. Let each actor go do a request for current items to the feed it is in charge of, and then send that list as an immutable object to the accumulating actor. The accumulating actor will maintain the aggregated list of feed items. When a new list of items is received from one of the other actors it merges that list into its aggregated list and at the end it has the full aggregated list.
Going a step further; why not have the accumulating actor send an immutable copy of the aggregated list to e.g. the view every tiume the aggregated list has been updated. That us allows us to show the user the partial results as they arrive.

Notice that the above did not involve any explicit starting of threads, no synchronization of threads, and no locking. Those things are handled behind the scenes by the actors and their message queues. Furthermore there is only mutable state in one place; inside the accumulating actor. Everything sent from one actor to another is immutable. So we've parallelized safely and easily.

The problem we're solving here is - admittedly - fairly easy to parallelize, but nonetheless I think the solution I described is nice because it is easy to understand and to code.

Why Do I Like Spec-Style Unit Tests?

As I've written about earlier I've been spending some time lately learning Scala. A side effect has been learning and using two test frameworks I haven't used before; ScalaTest and Scala Specs. Both offer what I think of as a spec style of writting tests. That is, tests along the lines of:

"A unit under test" should {
  "be tested by specs like this" in {
     // Code  for the first actual case
  }
 
  "and like this one" in {
    // Code for the second test case
  }
}

where a string is used to tell what's under test, and strings are also used to give each test case a name. In contrast "traditional" [J|N]Unit tests use method names to give each test case a name, and either don't point out the unit under test or use a class name for that.

The spec style of testing, I find, makes a test first workflow feel more natural, and makes my red/green cycles shorter, leading to more interaction between tests and code, and to better coverage. Very nice :-). I noticed this after using ScalaTest for a while, and found the same later when I started using Specs. Somehow both these frameworks gently nudge me towards a more clean TDD workflow.

I think the reason is that the unit under test is very explicitly named and that the test cases are named with strings rather than method names. Using a strings rather than method names makes me write longer names, and makes me write names that form full sentences when combined with the unit under test and the 'should' from the 'should' method. As in the example above which reads as "A unit under test should be tested by specs like this and like this one". The full sentences turn the tests into something a bit closer to specs, or rather, for me, the full sentences turn the tests into the place where I think about what exactly I want my classes to do, which makes it natural to write them first. On the other hand the natural language of full sentences makes me a little more aware of details, and therefore makes each test smaller and more focused which is what makes my red/green cycles shorter when using the spec style.

Smaller red/green cycles and a more test first approach has the usual benefits of turning the test into a design tool that leads to low coupling and high cohesion in the code under test, of creating high test coverage, and of catching coding bugs very early on.

That's why I like the spec style approach to unit testing.

First Adventures in ScalaTest - Part III: Introducing Actors in the NetWorth Sample

This is the third post about my little net worth Scala sample. This third post really doesn't have a lot to do with ScalaTest, so the title is probably somewhat misleading. What this post is about is how to introduce some parallelism in the net worth calculation code by using Scala actors. The first post in the series showed some test written in ScalaTest, the second one showed the net worth sample itself, and this one starts where that post ended: The code for calculating the net worth based on an XML file of stock symbols and units is:

val stocksAndUnitsXml = scala.xml.XML.load(fileName)

 val tickersAndUnits =
    (Map[String, Int]() /: (stocksAndUnitsXml \ "symbol")) {
      (map, symbolNode) =>
        val ticker = (symbolNode \ "@ticker").toString
        val units = (symbolNode \ "units").text.toInt
        map(ticker) = units
    }

 def generateTotalNetWorthAndSimpleReport = {
    val report = new java.io.ByteArrayOutputStream
    Console.withOut(report) {

      println("Today is " + new java.util.Date)
      println("Ticket  Units  Closing Price($)  Total value($)")

      val startTime = System.nanoTime

      val netWorth = (0d /: tickersAndUnits) {
        (cumulativeWorth, symbolAndUnitsPair) =>
          val (symbol, units) = symbolAndUnitsPair
          val lastClosingPrice = getLatestClosingPriceFor(symbol)
          val value = lastClosingPrice * units

          println("%-7s  %-5d %-16f  %-16f".format(symbol, units, lastClosingPrice, value))

          cumulativeWorth + value
      }

      val endTime = System.nanoTime

      println("Total value of investments is $" + netWorth)
      println("Report tool %f seconds to generate".format( (endTime - startTime)/1000000000.0 ))

      (netWorth, report)
    }
  }

That code is explained in my last post.

Now I want to parallelize that calculation. Specifically I want to parallelize the web service calls to get the latests price for each stock symbol, so I'm going to fetch each price in a separate actor and then send the prices back to the main thread and accumulate the net worth there.

The code that gets the price for a single symbol and sends it to another actor called 'caller' is:

    caller ! (symbol, getLatestClosingPriceFor(symbol))

To do that in a separate actor for each symbol I call this method:

   private[this] def getLatestPricesForAllSymbols(caller: Actor) =
     tickersAndUnits.keys.foreach {
       symbol =>
         actor {
           caller ! (symbol, getLatestClosingPriceFor(symbol))
         }
    }
  }

the call to 'actor' starts a new actor an executes the function given to it asynchronously. The '!' method on the actor 'caller' sends the value given as an argument to the '!' method to 'caller'.

To receive a single symbol and price I do:

  receiveWithin(10000) {
    case (symbol: String, lastClosingPrice: Double) =>
      val units = tickersAndUnits(symbol)
      val value = lastClosingPrice * units
  }

'receiveWithin(10000)' blocks until a message arrives for the current actor, or times out after 10 seconds, so if this code is in the 'caller' actor it will receive one of the symbol/price pairs sent using '!' in the code above.

In order to accumulate the whole net worth I place the above in a function that also takes an 'cucumulativeWorth' and returns an updated cumulative worth:

   private[this] def receiveAndProcessOneSymbol(cumulativeWorth: Double) = {
    receiveWithin(10000) {
      case (symbol: String, lastClosingPrice: Double) =>
        val units = tickersAndUnits(symbol)
        val value = lastClosingPrice * units

        println("%-7s  %-5d %-16f  %-16f".format(symbol, units, lastClosingPrice, value))

        cumulativeWorth + value
    }
  }

and to receive all the symbol/price pairs I call that method as many times as there are symbols in the 'tickersAndUnits' map:

  private[this] def receiveAndAccumulateWorthForAllSymbol =
    (0d /: (1 to tickersAndUnits.size)) {
      (cumulativeWorth, index) =>
        receiveAndProcessOneSymbol(cumulativeWorth)
    }

Putting all that together the 'generateTotalNetWorthAndSimpleReport' becomes:

  def generateTotalNetWorthAndSimpleReport = {
    val report = new ByteArrayOutputStream
    Console.withOut(report) {
      calculateNetWorthAndWriteReportTo(report)
    }
  }

  private def calculateNetWorthAndWriteReportTo(report: ByteArrayOutputStream) = {
      writeReportHeader

      val startTime = System.nanoTime
      getLatestPricesForAllSymbols(self)
      val netWorth = receiveAndAccumulateWorthForAllSymbol
      val endTime = System.nanoTime

      writeReportFooter(netWorth, endTime, startTime)

      (netWorth, report)
    }

  private def writeReportHeader = {
      println("Today is " + new java.util.Date)
      println("Ticket  Units  Closing Price($)  Total value($)")
    }

  private def writeReportFooter(netWorth: Double, endTime: Long, startTime: Long): Unit = {
    println("Total value of investments is $" + netWorth)
    println("Report took %f seconds to generate".format((endTime - startTime) / 1000000000.0))
  }

  private[this] def getLatestPricesForAllSymbols(caller: Actor) =
    tickersAndUnits.keys.foreach {
      symbol =>
        actor {
          caller ! (symbol, getLatestClosingPriceFor(symbol))
        }
    }

  private[this] def receiveAndAccumulateWorthForAllSymbol =
    (0d /: (1 to tickersAndUnits.size)) {
      (cumulativeWorth, index) =>
        receiveAndProcessOneSymbol(cumulativeWorth)
    }

   private[this] def receiveAndProcessOneSymbol(cumulativeWorth: Double) = {
    receiveWithin(10000) {
      case (symbol: String, lastClosingPrice: Double) =>
        val units = tickersAndUnits(symbol)
        val value = lastClosingPrice * units

        println("%-7s  %-5d %-16f  %-16f".format(symbol, units, lastClosingPrice, value))

        cumulativeWorth + value
    }

Now the calculation is done by fetching prices in parallel using the lightweight Scala actors, and sending the results back as one way messages that are aggregated to the final result in the main thread. That was easy, don't you think?

Oh, and finally even in this little sample and on a small dataset this actually does speed up things. If I rerun the integration test shown in the last post:

class IntegrationSpec extends FlatSpec with ShouldMatchers {
  def withYahooStockPriceFinder(fileName: String)(testFunctionBody: (PortfolioManager) => Unit) {
    testFunctionBody(new PortfolioManager(fileName) with YahooStockPriceFinder)
  }
 
  "A PortFolioManager with the YahooStockPriceFinder" should "produce a asset report and calculate the total net worth" in {
    withYahooStockPriceFinder("src/configuration/stocks.xml") {
      pm =>
        val (totalNetWorth, report) = pm.generateTotalNetWorthAndSimpleReport
        totalNetWorth should be(85000d plusOrMinus 10000)

        print(report)
    }
  }
}

I get:

Today is Wed Apr 28 20:37:38 CEST 2010
Ticket  Units  Closing Price($)  Total value($)
MSFT     190   30,930000         5876,700000    
INTC     160   23,220000         3715,200000    
ALU      150   3,150000          472,500000     
ORCL     200   25,940000         5188,000000    
NSM      200   14,970000         2994,000000    
CSCO     250   27,170000         6792,500000    
AMD      150   9,550000          1432,500000    
IBM      215   130,060000        27962,900000   
VRSN     200   26,840000         5368,000000    
XRX      240   10,980000         2635,200000    
APPL     200   0,000000          0,000000       
HPQ      225   53,330000         11999,250000   
ADBE     125   35,420000         4427,500000    
SYMC     230   17,270000         3972,100000    
TXN      190   26,380000         5012,200000    
Total value of investments is $87848.55
Report took 0,837941 seconds to generate

which shows that the calculation took less than a second now, whereas the calculation took over 6 seconds in the sequential version.
Oh, and also notice that the symbols appear in a different order in the report than they did in the last post. The input data is exactly the same. The difference is that they are printed in the order that the main thread receives prices for them, and that ordering is not deterministic any more because it depends on how fast each web service call returns.

And that's it. I'm still having fun with learning Scala :-)