Michael on development

Warning: do not try this at work! ;)

We were hosting a code retreat at Movio yesterday, and one of the sessions had the constraint that none of your production methods may return anything. I checked with our facilitator Richard and he said that callbacks are allowed.
So Mark and I decided that each method that would normally return a value, would instead take call a callback function with that value. We would hand in that callback into the method.
So we built a callback lifter that we would hand a function, which return value we’re interested in. It calls the function and returns the value provided in the callback. With this in place we can use any function as if it would return a value ;)

Here it is. Quite concise actually…

def callbackLifter[T](func: ((T => Any)) => Unit): T = {
  var retVal = None: Option[T]
  func((x: T) => retVal = Some(x))
  retVal.get
}

And here’s how to use is

def returnValueViaCallback(cb: (Int) => Any) { cb(3) }
callbackLifter(returnValueViaCallback) // 3

It also works with functions that take parameters

def withArgs(arg: Int, cb: (Int) => Any) { cb(arg) }
val partial = withArgs(99, _: (Int) => Any)
callbackLifter(partial) //99

We didn’t actually get very far in the kata, but this was big fun!
Let me repeat my warning: this was just a fun exercise to see what’s possible under funny constraints. Code retreats exist to drive some ideas to the extreme, which we did. ;)

In an adrenaline rush of good coding practices (I hope you read the sarcasm) we decided that mutable state in the callbackLifter (see the var?) is really bad. Exception driven development to the rescue! ;)

def exceptionDrivenCallbackLifter[T](func: ((T => Any)) => Unit): T =
  try {
    func((x: T) => throw new CallBackException[T](x))
    null.asInstanceOf[T] //the compiler forces us to return a T, even if the line below always throws an exception
  } catch {
    case CallBackException(x: T) => x
  }

At Movio we’re using Akka in more and more places – we love the Actor model for concurrency. This week we ran into a problem that I think more projects will face: we’ve got an Actor that produces work very quickly (i.e. it just pulls records from the database which is really fast) and sends that work as individual messages to some (routed) Actors who do some long-time processing on those. Our problem was that the producer is creating work much faster than all the worker Actors together will be able to work off. Those messages would all be waiting in the worker’s mailboxes, eventually eating up the complete heap space. I discussed this on the Akka usergroup and quickly got some good responses.

The solution we came up with is a simpler version of Derek Wyatt’s Balancing Workload Across Nodes with Akka that’s based on acknowledgements. We defined a master Actor who has access to some work – we called it Epic which is simply an Iterable[Work]. Worker Actors can run anywhere in your cluster and register with the master at any time. When the master get’s a new Epic it informs all workers that there’s work available. Those then pull piece by piece from the master until there’s no more work left. In the end the master simply doesn’t respond with work any more and the workers stop asking.

Your concrete implementation of this just needs to take care of the actual implementation of the worker (what does it actually do with that piece of work?), the Epic (how does work get created?) and the supervisor strategy of the Actors (in our case we made the workers restart and the master resume, as we don’t want it to loose the Epic it’s currently working on. Our Epic just pulls a batch of work from the database and queues it in a local queue. Once that queue is empty it refills it until there’s no more work left.

This fits nicely with the Actor model – nothing is blocking, you can distribute your workers and don’t need to worry about mailbox sizes. And here’s the important parts of the implementation. Our github repo contains the full implementation and the test suite. Simply clone it and run `sbt test`.

object WorkPullingPattern {
  sealed trait Message
  trait Epic[T] extends Iterable[T] //used by master to create work (in a streaming way)
  case object GimmeWork extends Message
  case object CurrentlyBusy extends Message
  case object WorkAvailable extends Message
  case class RegisterWorker(worker: ActorRef) extends Message
  case class Work[T](work: T) extends Message
}

class Master[T] extends Actor {
  val log = LoggerFactory.getLogger(getClass)
  val workers = mutable.Set.empty[ActorRef]
  var currentEpic: Option[Epic[T]] = None

  def receive = {
    case epic: Epic[T] ?
      if (currentEpic.isDefined)
        sender ! CurrentlyBusy
      else if (workers.isEmpty)
        log.error("Got work but there are no workers registered.")
      else {
        currentEpic = Some(epic)
        workers foreach { _ ! WorkAvailable }
      }

    case RegisterWorker(worker) ?
      log.info(s"worker $worker registered")
      context.watch(worker)
      workers += worker

    case Terminated(worker) ?
      log.info(s"worker $worker died - taking off the set of workers")
      workers.remove(worker)

    case GimmeWork ? currentEpic match {
      case None ?
        log.info("workers asked for work but we've no more work to do")
      case Some(epic) ?
        val iter = epic.iterator
        if (iter.hasNext)
          sender ! Work(iter.next)
        else {
          log.info(s"done with current epic $epic")
          currentEpic = None
        }
    }
  }
}

abstract class Worker[T](val master: ActorRef) extends Actor {

  override def preStart {
    master ! RegisterWorker(self)
    master ! GimmeWork // keep working on actor restart
  }

  def receive = {
    case WorkAvailable ?
      master ! GimmeWork
    case Work(work: T) ?
      doWork(work)
      master ! GimmeWork
  }

  def doWork(work: T)
}

For future reference here’s some alternative ideas – they all have caveats for the problem we faced, but might be considerable in other situations:

1. Have thousands of worker actors: doesn’t work for us because they depend on a database which is our actual bottleneck
2. Use a bounded mailbox size for the worker actors. That blocks the producing Actor when sending even more messages to the workers. Sounds like what we needed, however it doesn’t work with remote Actors: instead of blocking on a full mailbox it sends the message to the Deadletter Queue
3. Use Derek Wyatt’s PressureQueue. It’s basically a custom mailbox for the worker Actors that delays the submission of new messages based on the mailbox size. I’m not convinced that it fits the Actor model, partly because it’s blocking in producer Actor, which means that it can’t react on other messages any more. Also it doesn’t seem to be used widely and the last commit is 10 months ago.
4. The producer could only pull the Ids from the database and we hope that those fit into memory – i.e. we’d have millions of IDs as messages floating around. The workers then fetch the complete record later on and slowly get the job done. This works as long as all the Ids fit into memory – beyond that point your JVM explodes ;)
5. Use the TimerBasedThrottler which makes our producer only create X amount of work per time unit. The problem here is how do I get the X and the time unit? It’s only ever going to be a rough guess, so I’m either missing out on performance (if my workers could do faster) or potentially running out of memory (if my workers can’t catch up, e.g. because of other load on the system)
   

Scala 2.10 introduced a new reflection API (see here for an in-depth overview) which is much closer to the actual type system than the old Manifests were – in fact the compiler uses TypeTags itself. And it’s not hard to use as we’ll see. One reason why Scala provides reflection additional to Java’s reflection API is that the JVM erases generic types. As a motivation let’s see how generic types get erased at runtime:

  case class A[T]
  println(A[String].isInstanceOf[A[String]]) // true
  println(A[String].isInstanceOf[A[Int]])
  // true aswell - doesn't sounds right, no? that's because the type of T gets erased at runtime...

Scala 2.10 introduces TypeTag which brings us the compile-time information about the generic type T to the runtime:

  import scala.reflect.runtime.universe._
    case class B[T: TypeTag] {
      val tpe = typeOf[T]
  }
  println(B[String].tpe == typeOf[String]) // true
  println(B[String].tpe == typeOf[Int])    // false

  //and this also works for nested parameterised types:
  println(B[List[String]].tpe == typeOf[List[String]]) //true
  println(B[List[String]].tpe == typeOf[List[Int]])    //false

Another use case for the reflection API we just had at Movio is to get all members of an object that are of a given type. We define a customer specific configuration in objects that drives our whole system – from the Swagger API documentation all the way down to how we persist the data in Cassandra and do our filters. The code below grabs all members of an object that are subtypes of Field, gets their generic type T and checks if they mix in a given trait `Required`. Scala’s runtime mirrors allow us to lookup the symbols and types for a given instance at runtime – pretty neat!

  import scala.reflect.runtime.universe._
  case class Field[T: TypeTag] {
    val tpe = typeOf[T]
  }
  trait Required //can be mixed into Field to mark it as required

  object SomeClass extends BaseClass {
    val stringField = Field[String]
    val requiredIntField = new Field[Int] with Required
  }

  abstract class BaseClass {
    val typeMirror = runtimeMirror(this.getClass.getClassLoader)
    val instanceMirror = typeMirror.reflect(this)
    val members = instanceMirror.symbol.typeSignature.members
    def fieldMirror(symbol: Symbol) = instanceMirror.reflectField(symbol.asTerm)

    def fields = members.filter(_.typeSignature <:< typeOf[Field[_]])
    // filters all members that conform to type Field[_], i.e. also subclasses of Field
  }

  SomeClass.fields.foreach { symbol =>
    val name = symbol.name.toString.trim
    val required = symbol.typeSignature <:< typeOf[Required]
    val tpe = SomeClass.fieldMirror(symbol).get match {
      case field: Field[_] => field.tpe
    }
    println(s"field: $name; type=$tpe; required: $required")
    /** prints:
     *  field: requiredIntField; type=Int; required: true
     *  field: stringField; type=String; required: false
     */
  }

Credit goes to my colleague Felix Geller who discovered most of this first!