Make sure you get all your messages in your Scala code

01Jul09

I had this funny little Scala actor related bug today. Imagine you want to process a few things in parallel. So you go:

val processor = self
jobs.foreach { job =>
  actor {
    processor ! (job.id, job.run)
  }
}
// Merge results
for (i <- 1 to jobs.size) {
  self.receiveWithin(1000) {
    case (jobId:Int, result:JobResult) => mergeResult(result)
  }
}

All good. Then you realise that one or more of the jobs may fail with an exception. Which you have to handle somehow. So, you think, you’ll break on the first exception and report back. So you change that to:

val processor = self
jobs.foreach { job =>
  actor {
    try {
      processor  !  (job.id, job.run)
    } catch {
      case ex:Throwable => processor  !  (job.id, ex, job)
    }
  }
}
// Merge results
for (i <- 1 to jobs.size) {
  self.receiveWithin(1000) {
    case (jobId:Int, result:JobResult) => mergeResult(result)
    case(jobId:Int, ex:Throwable, job:Job) => throw new RuntimeException("Job " + jobId + " failed", ex)
  }
}

Cool. You fail on the first one – you just blow up and report back to your caller that something went wrong. Great! Well, not so…. Because the other jobs you started are still going to send you their results. You’ve stopped that thread by throwing an exception, so there’s nothing to receive those messages. When the web-server (in this case) reuses that thread, it will be sent all those messages. It won’t actually receive them until it hits the receiveWithin methond, so when you are expecting the return from the freshly started actors, you will actually be getting the messages from the actors that broke while servicing the last request. Kind of undesirable really.

One thing you can do is wait for all the messages. Take a response back regardless of what it is. This is what I did. Here’s the example:

 

      for (rowNumber <- 1 to mainRowsCount) {
        self.receiveWithin(2000) { 
           case (rowNumber:Int, subentities:Seq[Entity]) => {
              mainRowsWithJoins(rowNumber).addEntities(subentities)
           }
           case (rowNumber:Int, exception:Throwable, subrequest:Request) => {
              exceptions += new RuntimeException(”Subrequest ” + subrequest.toString + ” failed”, exception)
           }
        }
      }
for (jobNumber <- 1 to mainRowsCount) {

  self.receiveWithin(2000) { 

    case (jobNumber:Int, result:JobResult) => mergeResult(result)

    case (rowNumber:Int, exception:Throwable) => exceptions += new RuntimeException("Job " + job.toString + " failed", exception)

  }

}

Seems to cut the mustard for now, but I still think there’s something wrong with it – I don’t think you’re guaranteed to get all the messages, and I haven’t even started thinking about what will happen when I hit it with loads of requests running off the same Tomcat thread pool.

PS: Ok, now I’m reading this and thinking – that thread is dead, right, cause I blew it with the runtime exception. Except, it isn’t – some “self” is still receiving those messages when it says “receiveWithing”. Actors are not necessarily threads, I get that, and I’m sure it’s right, but it’s still quite confusing…..


Filed under: Uncategorized   |  Leave a Comment

Scala makes things expressive – timing

29Jun09

How often have you done something like this? -

public Result calculate() {
  long start = System.currentTimeMillis();
  Result myResult = doSomethingLongAndComplex();
  log.debug("Crunching took " + (System.currentTimeMillis() - start) + "ms.");
  return myResult;
}

It’s kind of ugly when you do it a few times and it does clutter the code. The need for defining an explicit myResult var is even more annoying. So, today, I added this to my Logging trait (see earlier post):

def timed[R](blockName:String)(block:=>R) = {
  val start = System.currentTimeMillis
  val result = block
  debug("Block (" + blockName + ") took " + (System.currentTimeMillis - start) + "ms.")
  result
}

What’s the big deal? – I hear you ask… The big deal is that I can say this in my code now:

def calculation : Result = {
  timed[Result]("calculate something") {
    doSomethingLongAndComplex
  }
}

And at the end of this block I’ll have the time in the log. Also, this block actually returns Result, so this could be the whole body of my function.


Filed under: Uncategorized   |  Leave a Comment
Tags: , , , ,

Scala makes things tidy – logging

29Jun09

Logging in Java can be messy. Especially the dreaded isDebugEnabled. Look – a class that calculates something:

 

import org.slf4j.LoggerFactory;
class SomeCalculator implements Calculator {
  private final log = LoggerFactory.getLogger(getClass());
  public Result caclulate(Input input) {
    if (log.isDebugEnabled()) {
      log.debug("Input is " + input);
    }
    Result result = input.times(64).shift(LEFT);
    if (log.isDebugEnabled()) {
      log.debug("Result is " + result);
    }
    return result;
  }
}

We have a class which has one line that actually does something and loads of lines that do logging. This is kind of simple, but in more complex cases it really messes your head up.


Now, consider this, in Scala:

 

class SomeCalculator extends Calculator with Logging {
  override def calculate(input:Input) : Result = {
    debug("Input is " + input)
    val result = input.times(64).shift(LEFT);
    debug("Result is " + result)
    result
  }
}

Why is this possible? Because I’ve made myself a nice little Logging trait, which looks something like this:

 

package com.supplesoftware.scala.log
import org.slf4j.LoggerFactory
trait Logging {
  val log = LoggerFactory.getLogger(getClass)
  def debug(msg: => String) = if (log.isDebugEnabled) log.debug(msg)
  ....
  def error(msg: => String, e:Throwable) = if (log.isErrorEnabled) log.error(msg,e)
}

So, as long as I add the logging trait to my Calculator class, I can just say debug(”something”). Note that the argument to debug() will not get evaluated unless it’s necessary – the argument is actually a function, not the actual evaluated string. So doing one isDebugEnabled in the actual Logging trait is fine and does not cause evaluation of the string.

In case that sounds weird, consider the debug method written like this:

def debug(msg: () => String) = if (log.isDebugEnabled) log.debug(msg())

 

I.e. the argument is not a by-name argument, but an actual function (which I guess is pretty much the same), taking no arguments and returning String. If debug is enabled, the VM will execute that function, to get the result – the log message.

Thanks for the folks at StackOverflow for the great suggestions.

Filed under: Uncategorized   |  Leave a Comment
Tags: , , ,

Fresh start

16Jun09

Having spent months pondering how to migrate my old blog to here, I though to myself, is it of any value to anyone? In this day and age, does anyone really need to generate hibernate mappings from xdoclet, or sneakily introduce open source technology through the back door, because the “architecture committee” won’t have it. That’s all in the stone age. If you still consider generating Hibernate XML mappings with xdoclet, then I probably don’t want you on my blog anyway.

There are of course old technology blog entries that are of value, but I don’t think mine were amongst them.

That, and the irony of not scrapping a blog that contains at least two articles on attachment, convinced me I should just let go of it. I mean, the blog is still there, just that all the old urls are broken. Might upset a couple of search engines, but, really, who cares?

I guess the lesson this has taught me is that it’s stupid to build your own thing when much better ones are available out there, often for free. I only built my own blog, because I wanted to try Ruby on Rails. It was fun doing it, but then kept becoming more and more high-maintenance, until I just stopped posting. While watching the free blog suppliers add more and more cool features, which I just couldn’t have, because I’d built my own and had my main business to do as well.

So here we are – blank sheet and all that :)


Filed under: Uncategorized   |  Leave a Comment