Monday, February 20, 2012

It's the familiarity model!

James Iry recently blogged on code density and tries to find out the meaning of "dense code". As an example he took upon the topic of regular expressions, which despite being dense are not frowned upon in coding and hardly get replaced for making the code fragment more readable.

So the question is what makes code dense so that it's not acceptable to programmers and they complain about it's incomprehensibility ?

Later in the blog post James himself identifies unfamiliarity as one of the culprits. People don't complain about regexes since they are familiar with them, but will surely complain of something else which they are not familiar with.

Almost during the same time Ola Bini blogged about expressiveness in programming language syntax. He mentions that a well designed syntax should help programmers *read* the code easily. But he also questions about the target programmers ..

who is this person reading ? It makes a huge difference if we’re trying to design something that should be easy to read for a novice or we’re trying to design a syntax that makes it easier for an expert to understand what’s going on.

Once again we get into this territory of familiarity and mental model. An expressive piece of code becomes readable only to a person who is familiar with the underlying model. In my programming career I have come across this dichotomy a number of times where programmers complain of something being too dense the moment it crosses the threshold of his familiarity level. I have seen developers taking every pain to understand the nuances of a Spring XML configuration. Or who have spent zillions of hours mastering the whole bunch of performance tuning Hibernate with stuffs like query cache configuration. Believe me it's not simple with tonnes of corner cases to take care of and even today I am not sure if it can be achieved in a deterministic way for all kinds of data models. But these same developers complain when they are faced with maintaining code that needs a basic understanding of functional programming, set theory or algebraic data types. I think it's purely because these form outside the limits of their familiarity model.

For a programmer who is not familiar with higher order functions, combinators like map, fold or filter will look too dense. So when you say map (+1) [1..5], the code fragment looks much less comprehensible to him than his familar variant of using an imperative mutated-indexed for-loop. To the unfamiliar the functional variant appears dense, to the expert it becomes succinct.

One of the challenges that I face today is to make programmers believe that learning new stuff will only help them think better. It's not mandatory that they will need all of these tools as part of their day job. But broadening your mental model can only help your thought process to leverage a wider playground. Maybe in our part of the world big companies give no incentive to transform yourself from a billable offshore resource to a thinking programmer. But you really need to transcend the limits of your familiarity model in order to appreciate code which experts certify as succinct.

Monday, February 06, 2012

Applicatives and a story of composability with sjsonapp


sjson has just gone applicative. I have changed the typeclasses for reading and writing jsons so that the typeclass protocols now return applicatives instead of raw types. Of course this makes the protocols composable with any other applicative based API in the world. This is the advantage of programming with generic abstractions like functors, applicatives and monads - you can compose them readily with any API that the world has written using the same ones.

Previously the serialization typeclasses for sjson looked like ..

// takes a type and produces a Json abstraction
trait Writes[T] {
  def writes(o: T): JsValue
}

// reads a json abstraction and produces T
trait Reads[T] {
  def reads(json: JsValue): T
}

You can compose writes and reads together only through function composition since they have symmetric type signatures. But you cannot get the benefits of threading additional effectful computations through them. e.g. I cannot accumulate errors generically in reads that result from mismatches in the field names between the json structure and the Scala type. Or I cannot plugin additional validations on Json structures during de-serialization into Scala type and have the validation messages passed on to the client. I need to have specialized handling in my code base by resorting to side-effects like throwing exceptions, which don't compose well.

In sjsonapp, the typeclasses are changed to ..

trait Writes[T] {
  def writes(o: T): ValidationNEL[String, JsValue]
}

trait Reads[T] {
  def reads(json: JsValue): ValidationNEL[String, T]
}

scalaz defines an applicative functor named Validation that allows you to compose validating abstractions. I had discussed in detail how you can compose domain models using applicative functors like Validation in an earlier post. You can use Validation to accumulate errors that occur when you de-serialize a Json structure into a Scala object.

Applicative Composition FTW

Here's the immediate impact of making your APIs return an applicative - your json processing becomes a composable pipeline of abstractions. Here's an example of an identity operation where you serialize Scala objects into json and de-serialize them back into the same abstractions using applicative composition ..

describe("Serialize and compose applicatives") {
  it("should compose and form a bigger ADT") {
    case class Address(no: String, street: String, zip: String)
    implicit val AddressFormat: Format[Address] =
      asProduct3("no", "street", "zip")(Address)(Address.unapply(_).get)

    case class Name(firstName: String, lastName: String)
    implicit val NameFormat: Format[Name] =
      asProduct2("firstName", "lastName")(Name)(Name.unapply(_).get)

    case class Me(name: Name, age: Int, address: Address)
    implicit val MeFormat: Format[Me] =
      asProduct3("name", "age", "address")(Me)(Me.unapply(_).get)

    val name = Name("debasish", "ghosh")
    val address = Address("1050/2", "Survey Park", "700075")
    val me = Me(name, 40, address)

    fromjson[Me](tojson(me).toOption.get) should equal(me.success)

    (tojson(name) |@| tojson(address) |@| tojson(40)) {(nm, add, age) =>
      (fromjson[Name](nm) |@| fromjson[Address](add) |@| fromjson[Int](age)) {(n, ad, ag) => Me(n, ag, ad)}
    } should equal(Success(Success(me)))
  }
}

Accumulating Validation Errors

Here's a test snippet that demonstrates how you can get back validation errors in a List when de-serializing a Json structure that's supposed to make a Scala object ..

case class Person(firstName: String, lastName: String, gender: String, age: Int)
implicit val PersonFormat: Format[Person] =
  asProduct4("firstName", "lastName", "gender", "age")(Person)(Person.unapply(_).get)

val pjson = """{"FirstName" : "Debasish", "LastName" : "Ghosh", "gender": "M", "age": 40}"""
fromjson[Person](Js(pjson)).fail.toOption.get.list 
  should equal(List("field firstName not found", "field lastName not found"))

The power of composition with applicatives .. and note we don't use any side-effecting operations like throwing exceptions which eschews purity of your functions. The accumulation is powered by a Semigroup abstraction that constrains the error part of the Validation. Have a look at how the applicative for Validation constrains X to be a Semigroup in scalaz and accumulates the failures in the last clause of the pattern match ..

implicit def ValidationApply[X: Semigroup]: Apply[({type λ[α]=Validation[X, α]})#λ] = 
  new Apply[({type λ[α]=Validation[X, α]})#λ] {
    def apply[A, B](f: Validation[X, A => B], a: Validation[X, A]) = (f, a) match {
      case (Success(f), Success(a)) => success(f(a))
      case (Success(_), Failure(e)) => failure(e)
      case (Failure(e), Success(_)) => failure(e)
      case (Failure(e1), Failure(e2)) => failure(e1 |+| e2)
    }
}

Besides mismatches in field names, you can also plug in custom validation functions that will be invoked during de-serialization of your json structures and report similar errors when they fail .. Here's an example ..

case class Person(firstName: String, lastName: String, gender: String, age: Int)

val validGender: String => ValidationNEL[String, String] = {g =>
  if (g == "M" || g == "F") g.success else "gender must be M or F".fail.liftFailNel
}

val validAge: Int => ValidationNEL[String, Int] = {a =>
  if (a < 0 || a > 100) "age must be positive and < 100".fail.liftFailNel else a.success
}

// the typeclass implementation for Person
implicit val PersonFormat: Format[Person] = new Format[Person] {

  // the de-serializing function
  def reads(json: JsValue): ValidationNEL[String, Person] = json match {
    case m@JsObject(_) =>
      (field[String]("firstName", m)            |@|
      field[String]("lastName", m)              |@|
      field[String]("gender", m, validGender)   |@| // validation plugin
      field[Int]("age", m, validAge)) { Person }
  
    case _ => "JsObject expected".fail.liftFailNel
  }
  
  // the serializing function 
  //..
}

Note how we plug in the validations for gender and age into the typeclass instance for Person. Also note that these functions also return an instance of scalaz Validation which can be nicely composed with the return type of the reads function after constructing the instance of the Person class. Here's an example ..

val p = Person("ghosh", "debasish", "M", 27)
fromjson[Person](tojson(p).toOption.get) should equal(p.success)

val r = Person("ghosh", "debasish", "G", 270)
fromjson[Person](tojson(r).toOption.get).fail.toOption.get.list should 
  equal(List("gender must be M or F", "age must be positive and < 100"))

Applicatives open up a world of possibilities

Once you have your APIs based on applicatives you can use all other abstractions that applicative functors offer - e.g. you can compose validations using Kleislis ..
describe("Serialize and chain validate using Kleisli") {
  case class Me(firstName: String, lastName: String, age: Int, no: String, street: String, zip: String)
  implicit val MeFormat: Format[Me] =
    asProduct6("firstName", "lastName", "age", "no", "street", "zip")(Me)(Me.unapply(_).get)

  val positive: Int => ValidationNEL[String, Int] =
    (i: Int) => if (i > 0) i.success else "must be +ve".fail.liftFailNel

  val min: Int => ValidationNEL[String, Int] =
    (i: Int) => if (i > 10) i.success else "must be > 10".fail.liftFailNel

  val max: Int => ValidationNEL[String, Int] =
    (i: Int) => if (i < 100) i.success else "must be < 100".fail.liftFailNel

  it("should serialize and validate") {
    val me = Me("debasish", "ghosh", 30, "1050/2", "survey park", "700075")
    val json = tojson(me)

    import Validation.Monad._
    type VA[A] = ValidationNEL[String, A]

    field[Int]("age", json.toOption.get,
      kleisli[VA, Int, Int](positive) >=> 
           kleisli[VA, Int, Int](min) >=> kleisli[VA, Int, Int](max)) should equal(30.success)

    val me1 = me.copy(age = 300)
    val json1 = tojson(me1)

    field[Int]("age", json1.toOption.get,
      kleisli[VA, Int, Int](positive) >=> 
           kleisli[VA, Int, Int](min) >=> kleisli[VA, Int, Int](max)).fail.toOption.get.list 
           should equal(List("must be < 100"))
  }
}
The new version of sjson with all the applicative based APIs is available in a separate repository sjsonapp on my github. Another interesting development that will soon be available is pluggable backend for sjson. The current version (branch master) uses dispatch as the json processing backend. I am working towards making sjsonapp compatible with RosettaJson, so that you can use pluggable json processing backends like LiftJson, Dispatch or BlueEyes. The current RosettaJson compatible version is available in the branch rosetta - feel free to checkout and play with it.

Tuesday, January 24, 2012

List Algebras and the fixpoint combinator Mu

In my last post on recursive types and fixed point combinator, we saw how the type equations of the form a = F(a), where F is the type constructor have solutions of the form Mu a . F where Mu is the fixed point combinator. Substituting the solution in the original equation, we get ..

Mu a . F = F {Mu a . F / a}

where the rhs indicates substitution of all free a's in F by Mu a . F.

Using this we also got the type equation for ListInt as ..

ListInt = Mu a . Unit + Int x a

In this post we view the same problem from a category theory point of view. This post assumes understanding of quite a bit of category theory concepts. If you are unfamiliar with any of them you can refer to some basic text on the subject.

We start with the definition of ListInt as in the earlier post ..

// nil takes no arguments and returns a List data type
nil : 1 -> ListInt

// cons takes 2 arguments and returns a List data type
cons : (Int x ListInt) -> ListInt


Combining the two functions above, we get a single function as ..

in = [nil, cons] : 1 + (Int x ListInt) -> ListInt

We can say that this forms an algebra of the functor F(X) = 1 + (Int x X). Let's represent this algebra by (Mu F, in) or (Mu F, [nil, cons]), where Mu F is ListInt in the above combined function.

As a next step we show that the algebra (Mu F, [nil, cons]) forms an initial algebra representing the data type of Lists over a given set of integers. Here we are dealing with lists of integers though the same result can be shown for lists of any type A.

In order to show (Mu F, [nil cons]) form an initial F-algebra we consider an arbitrary F-algebra (C, phi), where phi is an arrow out of the sum type given by :

C : 1 -> C
h : (Int x C) -> C


and the join given by [c, h] : 1 + (Int x C) -> C

By definition, if (Mu F, [nil, cons]) has to form an initial F-algebra, then for any arbitrary F-algebra (C, phi) in that category, we need to find a function f: Mu F -> C which is a homomorphism and it should be unique. So for the algebra [c, h] the following diagram must commute ..
which means we must have a unique solution to the following 2 equations ..

f o nil = c
f o cons = h o (id x f)


From the universal property of initial F-algebras it's easy to see that this system of equations has a unique solution which is fold(c, h). It's the catamorphism represented by ..

f: {[c, h]}: ListInt -> C

This proves that (Mu F, [nil, cons]) is an initial F-algebra over the endofunctor F(X) = 1 + (Int x X). And it can be shown that an initial algebra in: F (Mu F) -> Mu F is an isomorphism and the carrier of the initial algebra is (upto isomorphism) a fixed point of the functor. Well, that may sound a bit of a mouthful. But we can discuss this in more details in one of my subsequent posts. There's a well established lemma due to Lambek that proves this. I can't do it in this blog post, since it needs some more prerequisites to be established beforehand which would make this post a bit bloated. But it's really a fascinating proof and I promise to take this up in one of my upcoming posts. Also we will see many properties of initial algebras and how they can be combined to define many of the properties of recursive data types in a purely algebraic way.

As I promised in my last post, here we have seen the other side of Mu - we started with the list definition, showed that it forms an initial algebra over the endofunctor F(X) = 1 + (Int x X) and arrived at the same conclusion that Mu F is a fixed point. Or Mu is the fixed point combinator.

Sunday, January 15, 2012

Event Sourcing, Akka FSMs and functional domain models

I blogged on Event Sourcing and functional domain models earlier. In this post I would like to share more of my thoughts on the same subject and how with a higher level of abstraction you can make your domain aggregate boundary more resilient and decoupled from external references.

When we talk about a domain model, the Aggregate takes the centerstage. An aggregate is a core abstraction that represents the time invariant part of the domain. It's an embodiment of all states that the aggregate can be in throughout its lifecycle in the system. So, it's extremely important that we take every pain to distil the domain model and protect the aggregate from all unwanted external references. Maybe an example will make it clearer.

Keeping the Aggregate pure

Consider a Trade model as the aggregate. By Trade, I mean a security trade that takes place in the stock exchange where counterparties exchange securities and currencies for settlement. If you're a regular reader of my blog, you must be aware of this, since this is almost exclusively the domain that I talk of in my blog posts.

A trade can be in various states like newly entered, value date added, enriched with tax and fee information, net trade value computed etc. In a trading application, as a trade passes through the processing pipeline, it moves from one state to another. The final state represents the complete Trade object which is ready to be settled between the counterparties.

In the traditional model of processing we have the final snapshot of the aggregate - what we don't have is the audit log of the actual state transitions that happened in response to the events. With event sourcing we record the state transitions as a pipeline of events which can be replayed any time to rollback or roll-forward to any state of our choice. Event sourcing is coming up as one of the potent ways to model a system and there are lots of blog posts being written to discuss about the various architectural strategies to implement an event sourced application.

That's ok. But whose responsibility is it to manage these state transitions and record the timeline of changes ? It's definitely not the responsibility of the aggregate. The aggregate is supposed to be a pure abstraction. We must design it as an immutable object that can respond to events and transform itself into the new state. In fact the aggregate implementation should not be aware of whether it's serving an event sourced architecture or not.

There are various ways you can model the states of an aggregate. One option that's frequently used involves algebraic data types. Model the various states as a sum type of products. In Scala we do this as case classes ..

sealed abstract class Trade {
  def account: Account
  def instrument: Instrument
  //..
}

case class NewTrade(..) extends Trade {
  //..
}

case class EnrichedTrade(..) extends Trade {
  //..
}

Another option may be to have one data type to model the Trade and model states as immutable enumerations with changes being effected on the aggregate as functional updates. No in place mutation, but use functional data structures like zippers or type lenses to create the transformed object in the new state. Here's an example where we create an enriched trade out of a newly created one ..

// closure that enriches a trade
val enrichTrade: Trade => Trade = {trade =>
  val taxes = for {
    taxFeeIds      <- forTrade // get the tax/fee ids for a trade
    taxFeeValues   <- taxFeeCalculate // calculate tax fee values
  }
  yield(taxFeeIds ° taxFeeValues)
  val t = taxFeeLens.set(trade, taxes(trade))
  netAmountLens.set(t, t.taxFees.map(_.foldl(principal(t))((a, b) => a + b._2)))
}

But then we come back to the same question - if the aggregate is distilled to model the core domain, who handles the events ? Someone needs to model the event changes, effect the state transitions and take the aggregate from one state to the next.

Enter Finite State Machines

In one of my projects I used the domain service layer to do this. The domain logic for effecting the changes lies with the aggregate, but they are invoked from the domain service in response to events when the aggregate reaches specific states. In other words I model the domain service as a finite state machine that manages the lifecycle of the aggregate.

In our example a Trading Service can be modeled as an FSM that controls the lifecycle of a Trade. As the following ..

import TradeModel._

class TradeLifecycle(trade: Trade, timeout: Duration, log: Option[EventLog]) 
  extends Actor with FSM[TradeState, Trade] {
  import FSM._

  startWith(Created, trade)

  when(Created) {
    case Event(e@AddValueDate, data) =>
      log.map(_.appendAsync(data.refNo, Created, Some(data), e))
      val trd = addValueDate(data)
      notifyListeners(trd) 
      goto(ValueDateAdded) using trd forMax(timeout)
  }

  when(ValueDateAdded) {
    case Event(StateTimeout, _) =>
      stay

    case Event(e@EnrichTrade, data) =>
      log.map(_.appendAsync(data.refNo, ValueDateAdded, None,  e))
      val trd = enrichTrade(data)
      notifyListeners(trd)
      goto(Enriched) using trd forMax(timeout)
  }

  when(Enriched) {
    case Event(StateTimeout, _) =>
      stay

    case Event(e@SendOutContractNote, data) =>
      log.map(_.appendAsync(data.refNo, Enriched, None,  e))
      sender ! data
      stop
  }

  initialize
}

The snippet above contains a lot of other details which I did not have time to prune. It's actually part of the implementation of an event sourced trading application that uses asynchronous messaging (actors) as the backbone for event logging and reaching out to multiple consumers based on the CQRS paradigm.

Note that the FSM model above makes it very explicit about the states that the Trade model can reach and the events that it handles while in each of these states. Also we can use this FSM technique to log events (for event sourcing), notify listeners about the events (CQRS) in a very much declarative manner as implemented above.

Let me know in the comments what are your views on this FSM approach towards handling state transitions in domain models. I think it helps keep aggregates pure and helps design domain services that focus on serving specific aggregate roots.

I will be talking about similar stuff, Akka actor based event sourcing implementations and functional domain models in PhillyETE 2012. Please drop by if this interests you.

Sunday, January 08, 2012

Learning the type level fixpoint combinator Mu

I blogged on Mu, type level fixpoint combinator some time back. I discussed how Mu can be implemented in Scala and how you can use it to derive a generic model for catamorphism and some cool type level data structures. Recently I have been reading TAPL by Benjamin Pierce that gives a very thorough treatment of the theories and implementation semantics of types in a programming language.

And Mu we meet again. Pierce does a very nice job of explaining how Mu does for types what Y does for values. In this post, I will discuss my understanding of Mu from a type theory point of view much of what TAPL explains.

As we know, the collection of types in a programming language forms a category and any equation recursive in types can be converted to obtain an endofunctor on the same category. In an upcoming post I will discuss how the fixed point that we get from Mu translates to an isomoprhism in the diagram of categories.

Let's have a look at the Mu constructor - the fixed point for type constructor. What does it mean ?

Here's the ordinary fixed point combinator for functions (from values to values) ..

Y f = f (Y f)

and here's Mu

Mu f = f (Mu f)

Quite similar in structure to Y, the difference being that Mu operates on type constructors. Here f is a type constructor (one that takes a type as input and generates another type). List is the most commonly used type constructor. You give it a type Int and you get a concrete type ListInt.

So, Mu takes a type constructor f and gives you a type T. This T is the fixed point of f, i.e. f T = T.

Consider the following recursive definition of a List ..

// nil takes no arguments and returns a List data type
nil : 1 -> ListInt

// cons takes 2 arguments and returns a List data type
cons : (Int x ListInt) -> ListInt


Taken together we would like to solve the following equation :

= Unit + Int x a     // ..... (1)

Now this is recursive and can be unfolded infinitely as

= Unit + Int x (Unit + Int x a)
  = Unit + Int x (Unit + Int x (Unit + Int x a))
  = ...


TAPL shows that this equation can be represented in the form of an infinite labeled tree and calls this infinite type regular. So, generally speaking, we have an equation of the form a = τ where

1. if a does not occur in τ, then we have a finite solution which, in fact is τ
2. if a occurs in τ, then we have an infinite solution represented by an infinite regular tree

So the above equation is of the form a = ... a ... or we can say a = F(a) where F is the type constructor. This highlights the recursion of types (not of values). Hence any solution to this equation will give us an object which will be the fixed point of the equation. We call this solution Mu a . F.

Since Mu a . F is a solution to a = F(a), we have the following:

Mu a . F = F {Mu a . F / a}, where the rhs indicates substitution of all free a's in F by Mu a . F.

Here Mu is the fixed point combinator which takes the type constructor F and gives us a type, which is the fixed point of F. Using this idea, the above equation (1) has the solution ListInt, which is the fixed point type ..

ListInt = Mu a . Unit + Int x a

In summary, we express recursive types using the fix point type constructor Mu and show that Mu generates the fixed point for the type constructor just like Y generates the same for functions on values.

Sunday, January 01, 2012

2011 - The year that was

The very first thing that strikes me as I start writing a personal account of 2011 as it was is how it has successfully infused some of the transformations in my regular chores of programming world. It has been different and I am starting to enjoy some of the renewed vigor in areas like Type Systems, Machine Learning, Algebra etc. Throughout the year I used mostly one single language - Scala for programming with some occasional stints in Haskell and Octave for the Stanford Machine Learning course. But I have no regrets in not being more polyglotic, because I could find more time to dig deep into some of the more fundamental areas like algebra, category theory and type systems.

Favorite books read / started reading

  • Types and Programming Languages by Benjamin Pierce : definitely a Knuth statured book in the theory of type systems in programming languages. It's written with a very pragmatic outlook and contains all necessary implementation details to complement the accompanying theory. I have not yet finished reading the book. I am into Chapter 20 and 21 doing recursive types, that look to be one of the most exhaustive treatments of the subject I have ever seen. If and when I manage to finish reading this book, my next plan for theory of programming languages is Design Concepts in Programming Languages.
  • Conceptual Mathematics by F. William Lawvere and Stephen H. Schanuel - I started reading this book from recommendation by Paul Snively as a precursor to Benjamin Pierce's Category Theory for Computer Scientists. This is an excellent introduction to Category Theory and contains a detailed treatment of the unifying ideas of mathemetics, set theory and category theory.
  • Learn You a Haskell for Great Good by Miran Lipovaca - possibly the most recommended and updated Haskell reading in print form. The chapters on Applicative Functors, Monads and Zippers are real treats.
  • Language Proof and Logic by Jon Barwise and John Etchemendy - Starts with a great review of logic and goes on to discuss proofs of soundness and
  • A Tribute to a Mathemagician by Cipra, Demaine, Demaine and Rodgers - Another book in a series written by those illustrious mathematicians and puzzlers who were inspired by Martin Gardner. It's a fascinating collection of essays on mathematical puzzles - get it if you have that bent of mind.
Exploring new ideas

  • Category Theory - Often debated on its usefulness in the practical world, category theory gives you the basic understanding of programming language design, semantics and domain theory. I did lots of readings on Category Theory this year and this has led to a more concrete understanding of type systems as well. Hope to continue more in 2012.
  • Algebra - What's the algebra behind the term Algebraic Data Types ? I took some notes as I started understanding the algebra of recursive data types. Have a look at my notes on github.
  • Machine Learning - I took the Stanford online course on machine learning. It's been a revelation for me to find the pervasiveness of the subject in today's application context. Also the course encouraged me to look more into mathematics that govern all the theories that ML implements.
Some great papers read

Programming and Open Source

Once again a year passed by where I did 95% of programming in Scala. Scala has somehow hit the sweet spot of my liking - OO, FP, JVM, succinctness, I get them all in Scala. However, having said that I have every honest intention to renew all my friendships with Haskell and Clojure in 2012. I did quite a bit of Haskell in 2010 and still reaping the ebnefits of being a better Scala programmer piggybacking on my Haskell thoughts. I know Haskell is purer, a piece of Haskell code can be poetry. But the pragmatics of being on the JVM makes Scala more appealing to my professional life.

Two of my open source projects sjson and scala-redis are still quite active. I get pull requests on a regular basis and of course quite a few feature requests and bugs reported on Github. I plan to make some major upgrades to sjson particularly when reflection becomes more accessible in Scala 2.10. Also in line are some enhancements planned towards functor based JSON composition in sjson, which I plan to take up pretty soon. I tried to upgrade scala-redis to keep it in sync with the various releases of redis. Thanks to all of you for trying out sjson and scala-redis. Open source programming is fun and I consider myself blessed to have the opportunities to give something back to the community, which has given me so much over the years.

Any mention of my programming activities in 2011 would be incomplete without mentioning scalaz. I now use it in almost every project. It's really a great creation by Tony, Runar, Jason, Paul and the other members of the team. Using scalaz, I have learnt a lot about functional programming and functional thinking.

Another library that I have been using regularly since its inception is Akka. Asynchronous messaging is the gateway towards writing scalable applications and Akka provides the right set of batteries towards that. You get messaging, data flows, agents, STMs and all through a nice set of APIs both in Java and Scala. I think Akka is nicely poised to be the killer application to push Scala into the mainstream.

Some Publications

In 2011 I got the following two papers published, one of them as part of the esteemed team of Justin, Kresten and Steve. Thanks guys ..

  • Debasish Ghosh, Justin Sheehy, Kresten Krab Thorup and Steve Vinoski, "Programming Language Impact on the Development of Distributed Systems," FOME'11: Future of Middleware at Middleware'2011.
  • Debasish Ghosh, "DSL for the Uninitiated," Communications of the ACM, vol. 54, no. 7, pp. 44-50, July 2011
Some nice experiences

I attended 2 international conferences in 2011 - QCon London and PhillyETE. I also talked at PhillyETE on Domain Specific Languages. Both the conferences were amazing and I got to know in person many of the faces that I see and talk to regularly on Twitter and Google+. Incidentally I will also be talking at PhillyETE 2012 slated to be held in April.

My book DSLs In Action came out in late Dec 2010. 2011 was the year where I got the first royalty check from Manning. The writing of the book has been an amazing experience and to get to hear good words from people using the book gives another level of satisfaction. Thank you Manning for giving me the opportunity.

Looking forward to 2012

I am not one for resolutions, but here's a wish list towards more geekery in 2012 ..

  • Program more in Haskell and Clojure
  • Blog more (It was pathetic in 2011)
  • Do more math
  • Attend more online classes (currently registered for Natural Language Processing, Algorithms and Probabilistic Graphical Modeling at Stanford)
  • Try to do more conferences (currently registered for PhillyETE and Scala Days)
  • Learn more algebra, type theory and category theory
  • Get started with TAOCP Vol 4A
  • Learn Factor
Wish you a very happy new year. See you all in 2012!

Tuesday, September 27, 2011

Non blocking composition using Redis and Futures

scala-redis now supports pooling of Redis clients. Using RedisClientPool you can do some cool stuff in non blocking mode and get an improved throughput for your application.

Suppose you have a bunch of operations that you can theoretically execute in parallel, maybe a few disjoint list operations and a few operations on key/values .. like the following snippets ..

val clients = new RedisClientPool("localhost", 6379)

// left push to a list
def lp(msgs: List[String]) = {
  clients.withClient {client => {
    msgs.foreach(client.lpush("list-l", _))
    client.llen("list-l")
  }}
}

// right push to another list
def rp(msgs: List[String]) = {
  clients.withClient {client => {
    msgs.foreach(client.rpush("list-r", _))
    client.llen("list-r")
  }}
}

// key/set operations
def set(msgs: List[String]) = {
  clients.withClient {client => {
    var i = 0
    msgs.foreach { v =>
      client.set("key-%d".format(i), v)
      i += 1
    }
    Some(1000) // some dummy
  }}
}

Redis, being single threaded, you can use client pooling to allocate multiple clients and fork these operations concurrently .. Here's a snippet that does these operations asynchronously using Scala futures ..

// generate some arbitrary values
val l = (0 until 5000).map(_.toString).toList

// prepare the list of functions to invoke
val fns = List[List[String] => Option[Int]](lp, rp, set)

// schedule the futures
val tasks = fns map (fn => scala.actors.Futures.future { fn(l) })

// wait for results
val results = tasks map (future => future.apply())

And while we are on this topic of using futures for non blocking redis operations, Twitter has a cool library finagle that offers lots of cool composition stuff on Futures and other non blocking RPC mechanisms. Over the weekend I used some of them to implement scatter/gather algorithms over Redis. I am not going into the details of what I did, but here's a sample dummy example of stuffs you can do with RedisConnectionPool and Future implementation of Finagle ..

The essential idea is to be able to compose futures and write non blocking code all the way down. This is made possible through monadic non-blocking map and flatMap operations and a host of other utility functions that use them. Here's an example ..

def collect[A](fs: Seq[Future[A]]): Future[Seq[A]] = { //..

It uses flatMap and map to collect the results from the given list of futures into a new future of Seq[A].

Let's have a look at a specific example where we push a number of elements into 100 lists concurrently using a pool of futures, backed by ExecutorService. This is the scatter phase of the algorithm. The function listPush actually does the push using a RedisConnectionPool and each of these operations is done within a Future. FuturePool gives you a Future where you can specify timeouts and exception handlers using Scala closures.

Note how we use the combinator collect for concurrent composition of the futures. The resulting future that collect returns will be complete when all the underlying futures have completed.

After the scatter phase we prepare for the gather phase by pipelining the future computation using flatMap. Unlike collect, flatMap is a combinator for sequential composition. In the following snippet, once allPushes completes, the result pipelines into the following closure that generates another Future. The whole operation completes only when we have both the futures completed. Or we have an exception in either of them.

For more details on how to use these combinators on Future abstractions, have a look at the tutorial that the Twitter guys published recently.

implicit val timer = new JavaTimer

// set up Executors
val futures = FuturePool(Executors.newFixedThreadPool(8))

// abstracting the flow with future
private[this] def flow[A](noOfRecipients: Int, opsPerClient: Int, fn: (Int, String) => A) = {
  val fs = (1 to noOfRecipients) map {i => 
    futures {
      fn(opsPerClient, "list_" + i)
    }.within(40.seconds) handle {
      case _: TimeoutException => null.asInstanceOf[A]
    }
  }
  Future.collect(fs)
}

// scatter across clients and gather them to do a sum
def scatterGatherWithList(opsPerClient: Int)(implicit clients: RedisClientPool) = {
  // scatter
  val allPushes: Future[Seq[String]] = flow(100, opsPerClient, listPush)
  val allSum = allPushes flatMap {result =>
    // gather
    val allPops: Future[Seq[Long]] = flow(100, opsPerClient, listPop)
    allPops map {members => members.sum}
  }
  allSum.apply
}

For the complete example implementations of these patterns like scatter/gather using Redis, have a look at the github repo for scala-redis.