Java to the Limit

Erlang is not a Concurrent Functional Programming Language

Whenever developers fall in love with Erlang (as many tend to do these days) they are too often busy promoting it to their friends and colleagues as a cool language for concurrency and functional programming. Which is a pity, for several reasons.

1. It intimidates people.
2. It's wrong.

The intimidation factor

Functional programming is scary. Most developers for whom Erlang may be a relevant alternative are probably doing server-side programming in Java or C# today: they most likely know by experience that concurrent programming is damn hard, and functional programming was something they tried +10 years ago in a computer science class in their undergraduate work. So if you focus on those two aspects you are almost certain to intimidate people, and convince them that this is really difficult stuff to lean.

As we all know ... "people don't want to buy quarter inch drills, they want quarter inch holes." Likewise with programming languages: If you want to advocate Erlang to your friends and colleagues, focus on the benefits. It's not really a key selling point of Erlang that it has certain resemblances with some functional programming languages, neither is it a primary quality that it is really easy to spin up a million processes. Those are good-for-nothing arguments, that just sound flashy, and focus on some secondary aspects of the artifact called Erlang.

Focusing on concurrency also has the problem that most people don't even know what it means, nor why it is important. On blogs and online media there is a lot of discussion about on issues with utilizing multi-core CPUs, the free lunch is over, etc. etc. Combine that with your average engineer's performance fetich, and many people will intuitively think that the concurrency is there to get performance. Why else would you bring such mind-boggling pains upon yourself?

Being wrong

When I started looking into Erlang two years ago, I was wrong. I was looking for something to make concurrent asynchronous programming more intuitive, because those things were (and still are) obviously hard in the Java world I'd been living in for the last 15 years. It is difficult to write programs that utilize multicore; but in many cases -- I've realized -- that is of lessor importance. I went looking for one thing, but found something much more important.

Erlang is about building reliable systems.

That's it: Erlang is about making software systems that are fault tolerant, given that just about everything about the system has problems: hardware and networks fail (often because of human err), the software we write has bugs that need fixing, requirements change all the time, and we want the system to run 24x7.

You can almost hear the designers saying: "Let's make a tool to build reliable systems; what do we need to make that happen?" Then then they went off and solved all the hard problems involved with that, and to this date, Erlang is the only "full package" that can do that. And as complete packages go, there are some rough edges and individual things that could be made better, but ... as a whole it's pretty damn good. And don't get me wrong: Software is not automatically reliable because it is written in Erlang ... rather, Erlang is a toolbox that allows you to write reliable software systems.

What is Erlang then?

First of all, Erlang introduces mechanisms to contain and manage faults. When a problem happens in an Erlang system, it doesn't take down everything with it. This is hard to grasp for a Java programmer where you're used to a bad null pointer error or something that will take down the entire application. Erlang is more like an operating system which itself runs inside an OS-level process: if one component faults, then its resources are properly reclaimed and only that component is shut down.

To manage faults, Erlang has mechanisms much like Unix signal handlers, that will notify other components about faults. These are called links and monitors. Monitors are one way (used to observe), whereas links are bidirectional, and can be used to establish strong aggregation-structures of Erlang components so that they "die in unison": if one goes down the integrity of the others is also at risk.

In order to make a system reliable, you also need to support running "the system" on multiple computers. To this end, Erlang has transparent distribution so you can have a component one one physical machine monitoring (or being linked to) a component on another machine. If an entire (virtual) machine dies, components on the other machine will observe that as if all the individual components on the failing machine faulted.

The process abstraction is used to capture all of the above. I said "component" but that really is process, which is something we all know well already. In Erlang, processes don't have threads, and processes can't see or touch each other's data. Just like in an operating system process. Erlang processes are individually sequential; and so an Erlang system is really just a set of Cooperating Sequential Processes. It reduces the inherent concurrency to just a bunch of sequential programs. If you want, you can think of processes as actors; which is really just how objects should have been in the first place. Actors encapsulate their state, and they use message passing to interact.

Further, you need to have a model for fixing bugs while the system is running. That's the key feature of Erlang, The Movie. At its core, Erlang has features to support upgrading code in flight, which is also one of the reasons why Erlang is not a statically typed language: if you need to change your data structures in flight, you also need to change the types. While code upgrade is something which is very handy for development, it is not at all easy to do properly in production. I've heard that for some of Ericsson's telephony systems where they do live upgrades, they spent as much time developing and testing the upgrade as they do developing and testing new features. The frameworks and libraries that come with Erlang help you in many cases, as long as you stay within certain limits.

But perhaps most importantly, Erlang is just plain simple. Erlang code is compact, and (once you get used to it) very easy to read. Compared to coding Java, it almost feels like cheating.

So to get all these goodies, you have to learn some new syntax. Get over it.

Before you jump to conclusions and barf at me in the comments below, keep in mind that I do know Erlang very well. I've written an alternative Erlang VM, and I was awarded "Erlang User of the Year 2010". So, I do know that Erlang is a concurrent and somewhat functional programming language. But that's not the point.

Using Links to Index Riak objects

Riak doesn't come with means to do secondary indexes out of the box. This limits many applications of Riak; and it's a bit tricky to implement correctly oneself.

The module riak_link_index provides a mechanism that can be applicable for some situations, providing means to create synthetic (secondary link index) objects triggered by a Riak commit hook. The mechanism work by defining a function that names these synthetic objects as a function of the primary object body.

Basic Example

Assume the primary object contents being stored under the name dogs/ABC01 is

{
  type: "Dog",
  name: "Fido",
  owner: "Peter"
  ...
}

And the indexing function looks like this:

function indexDogs(metaData, contents) {
   Body = JSON.parse(contents);
   if (Body.type == "dog" && Body.owner) {
       return [ ["people", Body.owner] ];
   } else {
       return []
   }
}

Then a secondary index object will be created named /riak/people/Peter, which contains a Riak link pointing back to dogs/ABC01.

That's it, basically. There are a few extra details though.

Composability by Tagging

Indexes are virtual entities not tied to a particular bucket; but the indexing function needs to be "installed" in the bucket being written to. However, multiple buckets may have objects that get indexed the same way, and the "index buckets" (in the above case the /riak/owns bucket) may contain different kinds of indexes. This kind of composability is controlled by assigning a unique tag to each indexer.

When you install an indexer, you also name it's tag. It could be that in the above example you choose the tag "owns_dog", then you can use a link-walk query like this:

curl http://127.0.0.1:8091/riak/people/Peter/_,_,owns_dog

To return all the dogs that are owned by peter. But the people/Peter object can also contain links to objects that are indexed by another indexer with it's own tag; say the houses that people own.

Under the hood

What really happens in the above example is that the indexer function runs in the precommit phase and creates links attached to the primary object, pointing to the index objects, tagged with idx@owns_dog. So, the object dogs/ABC01 will get a link

Link: </riak/people/Peter>; riaktag="idx@owns_dog"

The post-commit phase of storing the dog object, will then go to the targets of all the idx@-prefixed links (the index objects), and make sure that those each have a link pointing back to the dog.

So, in stead of using the indexer function, you can also just provide those links when you push the object in the first place; and then the indexer function doesn't have to be installed.

The index objects contain a data structure used to ensure the consistency of the resulting links; that is if two different PUTs to some primary objects mutate the same index object, then the index object may end up having a write conflict (siblings), and the data structure makes sure that such conflicts are resolved properly with it's own set of vector clocks.

Because of that, you cannot use the index objects to contain other stuff. The data structure contained therein is only for the use of the riak_link_index stuff.

Introducing Vector Maps

This post introduces a new data structure - the vector map - which solves some issues related to storing collections in MVCC data stores. Further, vector maps have some super nice use cases for "occasionally connected" systems.

The idea warrants a more rigorous discourse, but I need to get it off my chest, so here is a blog entry describing it.

Modern distributed data stores such as CouchDB and Riak, use variants of Multi-Version Concurrency Control to detect conflicting database updates and present these as multi-valued responses.

So, if I and my buddy Ola both update the same data record concurrently, the result may be that the data record now has multiple values - both mine and Ola's - and it will be up to the eventual consumer of the data record to resolve the problem. The exact schemes used to manage the MVCC differs from system to system, but the effect is the same; the client is left with the turd to sort out.

This led me to an idea, of trying to create a data structure which is by it's very definition itself able to be merged, and then store such data in these kinds of databases. So, if you are handed two versions, there is a reconciliation function that will take those two records and "merge" them into one sound record, by some definition of "sound".

From what I have seen, the "thing" stored is often itself a collection like a list or a hash map, and say that Ola and I both add new elements to the collection and store the results, the resulting multiple records are - with proper definitions - naturally mergeable; namely the list or map that contains the original entries plus both mine and Ola's.

So, this is the presentation of my idea: A vector map, which is a data structure that is designed to be used in this context. It also has other interesting applications as we shall discuss towards the end of this post.

Vector Maps

A vector map is defined as a set of assignment events, Key=Value, each such event being time stamped with a vector clock (hence the name). From a high-level point of view, a vector map can be seen as a hash table i.e., a collection key/value pairs.

Two vector maps can be reconciled (VectorMap₁ ⊕ VectorMap₂), so that for each key, the "most recent assignment event" wins. If assignment events are in conflict (vector-clock wise concurrent), then the resulting value is multivalued.

The reconciliation function for vector maps is defined so that it is commutative i.e., it can be applied in any order i.e.,

A ⊕ B = B ⊕ A,

It is also associative,

A ⊕ (B ⊕ C) = (A ⊕ B) ⊕ C,

which means that reordering done inside the database store, is insignificant to the resulting usage.

In fact, Riak itself (the distributed key/value store) is just a big distributed and redundant version of this; which just goes to prove the versatility of the idea. So in a sense, a vector map is just a recursion over the Riak (Dynamo) concepts applied to a data structure.

In the following we will define things a little more rigorously and provide some examples; and towards the end there is a discussion of how vector maps can be used.

Assignments as Vector Clocked Events

For the purposes of this discourse, we will model an assignment event as

• a Key,
• a Set of values, and
• a vector clock, VC, describing when the assignment happened,

so an assignment event has the following form:

Assignment :: VC: Key = [Value, Value, ...]

We will let the set contain multiple values in case a conflict has been observed, but to start off with our sets will be single-valued.

Our system will use the vector clock to determine "who wins". Let's see what happens ...

To begin with, I do an assignment "X" = 4, and record this event:

Assignment₁ = (krab:1) : "X" = [4]

Reading: at krab's time 1, "X" is bound to 4.

Later, after observing my assignment, a colleague Ola, at his time 2, defines "X" to be 5

Assignment₂ = (ola:2,krab:1) : "X" = [5]

Now, the immediate beauty is that because each assignment is time stamped with a vector clock, we can easily determine that Assignment₂ happened after Assignment₁ (the vector clock says that Ole did indeed observe Assignment₁ before creating his own), and so if we see both we can discard the earlier one without loss.

Conflict happens

Now, what happens if Jens comes in, and based on only observing my original assignment, he reassigns "X" to be 7. We'll describe this with the following event:

Assignment₃ = (jens:3,krab:1) : "X" = [7]

To Jens, this is not problematic, but if someone observes both Assignment₂ and Assignment₃ they'll know that Jens did not observe Ola's time 2.

To make some kind of sense out of this, we define the reconciliation operator ⊕, describing the aggregation of knowledge we have when combining the known events.

Assignment₂ ⊕ Assignment₃ = (krab:1,ola:2,jens:3) : "X" = [5,7]

I.e., we describe that "X" has conflicting values 5 and 7 at a point in time which is after both Assignment₂ and Assignment₃.

Deleting values

A further complication is what happens when we want to delete a binding, but this is handled quite simply by making a new assignment to a unique tombstone value which is somehow outside scope of other possible values; and then reconciling that into the current event set. This does have some interesting properties, because there is an observable difference between a binding that was never there, and a binding that was removed.

Defining Reconciliation

In the example above, we can see from the vector clocks that Jens had not observed Ola's assignment, and so we reconciled the two conflicting events by

1. creating an artificial vector clock (krab:1,ola:2,jens:3) which is logically after both (ola:2,krab:1) and (jens:3,krab:1), and
2. combining the bound values [5] and [7] into a set of values [5,7].

Had it been obvious that one assignment happened before the other, we would simply had thrown one of them away, but because the two vector clocks were in conflict we have to recognize the conflict.

So the actual definition of reconciliation for two assignment events with the same key is as follows:

VC₁:Key=Set₁ ⊕ VC₂:Key=Set₂ ≡
        VC₁ ≤ VC₂ → VC2 : Key = Set₂;
        VC₂ ≤ VC₁ → VC₁ : Key = Set₁;
        otherwise → lub(VC₁,VC₂) : Key = (Set₁ ∪ Set₂)

The reconciliation operator (o-plus) is commutative (just like good old addition), so we can use it to reconcile assignment events in any order and we'll always arrive at the same result in the end. Which makes it perfect for making decisions in a distributed system; because it means that even if we get to know about events in different order, the state will eventually reconcile to the same value (eventually meaning when we've seen all the events).

The definition uses the lub (least upper bound) on two vector clocks, which combines two such by taking the maximum local time stamp for each agent present in two sets of vector clocks:

lub( VC₁, VC₂ ) ≡
        ∀ a ∈ agents(VC₁) ∪ agents(VC₂)
            a : max( time(VC₁,a), time(VC₂,a) )

Where time(VC,a) is 0 (zero) if VC does not list an agent a. For example

lub( (b:3), (a:1, b:2) ) = (a:1, b:3)

What is this good for?

You may rightly say that this doesn't solve the problem, it just pushes the problem one level down. And that's right; but ... it does solve an interesting range of problems, and with some care you can often structure your usage of keys in vector maps so that you can avoid conflicts all together.

Further, since your favorite data store already does this for you ... you may say that you can just split your map into individual key/value bindings and store those in the database. But that comes at a price of network round trips and data locality.

Use case: Modeling Relationships in MVCC data stores

Vector maps are really nice for modeling relationships inside MVCC databases.

Assuming you want to store a one-to-many relationship in Riak (say, order - order-item), you run into the problem on the one-side that it is likely to be concurrently updated if multiple items are added to the same order.

With vector maps, you can easily model the entire relationship as one order object which is just a vector map, where each item is stored with a distinct key. If that key is e.g. a sufficiently large random number, making it very unlikely that order-item-id's conflict, then you're pretty much home free. Alternatively, you can devise a mechanism so that each client of the system is able to construct globally unique keys (agent + sequence number).

For this use case, you'll see much improved performance also, because of the improved locality of reference. If your data store needs to go and fetch each individual order-item on the disk somewhere, then performance will be seriously hampered.

Use case: off-line data

Vector maps are great for off-line data, because they give a well defined meaning to the concept of synchronization (something I would really like my iCal to do :-) Synchronization is simply defined as the exchange of vector maps, storing the result of the reconciliation on both sides.

Such synchronization can happen in near-real time (peer-to-peer update) or as a delayed synchronization whenever there is contact to a server/peer.

This is perhaps the most interesting use case, because it an be used as a simple foundation for making data available to e.g. mobile clients in an "occasionally connected" system, in a way that makes sense for both online and offline mode.

Implementation Issues

Since it is a little complicated to manipulate a vector map, we need implementations in the most common languages out there to get it off the ground. I'm currently hacking on an Erlang and a Java version.

Working on this, and I have come to the conclusion that it would be great if vector maps have a well defined binary representation so that they can be meaningfully manipulated in a number of different contexts, easily stored and transmitted, etc. So, a special mime-type that lets multiple parties consume vector maps independent of programming languages.

If vector maps had it's own mime type and well defined data representation, data stores such as Riak or CouchDB could even do the reconciliation automatically before serving the data to a client.

So, right now I am working with a protocol buffers definition that looks like this on the wire, to be encoded as Content-Type application/x-protobuf;proto=vectormap, and likely a also a JSON representation application/json;schema=vectormap.

message VectorMap {
   repeated Entry entries    = 1;
}

message Entry {
   required string key       = 1;
   repeated Clock  vclocks   = 2;
   repeated Value  value     = 3;
}

message Value {
   optional string mime_type = 1 [ default = "application/json;charset=utf-8" ];
   optional bytes  content   = 2;
   optional bool   deleted   = 3 [ default = false ];
}

message Clock {
  required string node       = 1;
  required uint32 counter    = 2; 
  required uint64 utc_millis = 3;
}

Conclusions

The Dynamo idea of using vector clocks to time stamp data is great, but I think the power of the idea goes quite a bit further if the logic is brought all the way to the client.

CouchDB tries to do this by suggesting that client devices (mobile devices) should have a full fledged CouchDB running there. But I think that exposing data this way makes the mobile client model much more manageable. This entire idea can be wrapped up in a single Java class which is easily deployed in a Android app; and it is sufficiently simple to be implementable in a range of programming languages.

What do you think?

About Version Vectors (a.k.a. Vector Clocks)

The last few weeks I've been working on understanding vector clocks (or rather, version vectors). When working with distributed data stores like Riak and Voldemort which are using this to manage consistency, it is important to have a deep understanding of what kind of consistency they provide, and how to design systems that work well with these structures.

So, without further ado; here's my story. We'll start somewhere else though.

Versions of shared spreadsheets

Adam and Eve are working on a shared Excel spreadsheet, and they have devised a clever scheme for versioning it, so they can detect conflicting edits.

Their idea is that the document has two version numbers embedded in the file name. When Adam saves a new version, he increments his number; and likewise for Eve. After some exchanges and edits, the document is now called doc_adam3_eve4.xls (Adam in version 3, Eve in version 4). The rules of the game are thus:

• If Adam updates the document, he shall increment from adam3 to adam4, and so name the next version doc_adam4_eve4.xls, whereas
• if Eve updates the same document she shall increment eve4 to eve5 and call the result doc_adam3_eve5.xls.

Because both Adam and Eve have their own version scheme, they can make edits and when they share their updated documents they can easily detect if they have made concurrent edits. Eve will bring her maximum eve-numbered version, and Adam will bring his maximum adam-numbered version. Then a mechanism depicted in this diagram can be used to detect a conflict:

Given version {adam=3,eve=4} (represented by the white square in the middle), you can see which versions are logially "after" and "before" that version. Versions colored yellow are in conflict with doc_adam3_eve4.xls, representing logically "concurrent" edits.

The great thing is that this scheme extends to any number of participants. It's just easier to depict in a two-dimensional way, but the same mechanism works for any number of vector dimensions. Missing participants are assumed to have version 0.

Vector Time. The numbering scheme above is an example of a vector clock which can be used to describe temporal relations between events in a distributed system. Generally speaking, a vector clock is a list of (place, version) pairs, in which each place occurs at most once.

Really, we should perhaps talk about version vectors because the time-notion does not exactly refer to our intuitive notion of time, but to some monotonous increasing series of "versions" issued at each "place".

version vectors and version vectors have a partial ordering, which captures the idea of before/after a given point in vector time. It's partial because there exists pairs of version vectors so that one is neither before nor after the other; they are "concurrent".

If you have a set of version vectors, there is thus a subset hereof which are "maximal" (most recent). This set contains only one element if there is no conflict.

If you use version vectors to give versions to documents, then it is obviously great if there is only one such "most recent" document. If there is more than one "most recent" then at least you have narrowed it down to the versions to take a closer look at.

Assembling the Jigsaw

Regardless of their cleverness, Adam and Eve are often in trouble, because often they have more than one "most recent" Excel spreadsheet when they get together (say, doc_adam4_eve4.xls and doc_adam3_eve5.xls mentioned before), and so they have to sit there and take their spreadsheet apart, resolve the conflicts, and put it back together. Once they have resolved their conflicts they can name the new version doc_adam4_eve5.xls which is "after" both of the aforementioned ones and be back on track.

Now Justin (he's a really smart programmer) comes along to help them write a new spreadsheet application called Schemix that solves their issues. Schemix represents spreadsheets such that each cell value in the spreadsheet has it's own version number using version vectors.

Here's a document in Schemix

And here is how Schemix represents this data after Adam and Eve have been working on it for a while.

A1, ["Anders"],                 { adam:3, eve:4 }
B1, ["anders@somewhere.com"],   { adam:1        }
A2, ["Brandon"],                { adam:2, eve:5 }
B2, ["brandon@elsewhere.com"],  {         eve:4 }

The right-most column is the version vector timestamp for the corresponding cell. So, in stead of giving the entire document a single version, each cell in the spreadsheet has its own independent version number. Whenever Adam saves his document, the cells he has modifies will have their version number bumped up one.

This versioning scheme allows any two such Schemix documents to be merged, by comparing the cells individually using the scheme Adam and Eve were using above. Assume the aforementioned document was merged with this:

A1, ["Andy"],                   { adam:4, eve:3 }
B1, ["anders@somewhere.com"],   { adam:1        }
A2, ["Bill"],                   { adam:3, eve:5 }
B2, ["brandon@elsewhere.com"],  {         eve:4 }

Merging two Schemix documents is like assembling a jigsaw puzzle from two sets of jigsaws puzzle pieces:

• For some pieces, there will be two identically versioned ones (such as cells B1 and B2) and so you just choose any one and throw the other away.
• For some pieces, one is "newer" than the other (for cell A2); and so you just keep the most recent one and discard the other.
• but, but, but for some there may be a conflict, and so the cell becomes multi-valued. Say, both Adam and Eve concurrently change the cell A1, but to distinct values. Then you're in trouble; but luckily much less so than before because the conflict only pertains to that single cell.

The merged document look like this:

A1, ["Andy", "Anders"],         { adam:4, eve:4 }
B1, ["anders@somewhere.com"],   { adam:1        }
A2, ["Bill"],                   { adam:3, eve:5 }
B2, ["brandon@elsewhere.com"],  {         eve:4 }

When cells are in conflict as a result of merging two Schemix documents, they are allowed to have multiple values. In Schemix, this is obviously just represented as a multi-values GUI element (a drop-down), so that once you have a merged document, you can easily edit it to resolve the conflict.

Inremental Updates

Version vectors are really nice, because you don't need entire documents to be able to do consistent merges. If Adam and Eve can see each other via a peer-to-peer channel, they can just as well send updates to each other by sending the cell values incrementally. The fact that each cell carries it's own consistency structure makes this scheme applicable to bulk- or incremental merges alike.

Relation to NoSQL

As mentioned above, this idea of version vectors is important because it is a fundamental notion underlying many of the new distributed NoSQL data stores, such as Voldemort, and Riak. The cell-names A1, B2, and so on are similar to the keys used in these systems, and the notion of version vectors described here is close to what they do. A Riak store is really just one giant version of the Schemix spreadsheet I described above.

The interesting thing is that even though they do provide automatic reconciliation, there is no guarantee about the referential integrity of the result. Two independent edits to two distinct key/values can always be merged; but they could easily break some external invariant which is not expressed explicitly.

What kinds of strategies should we then employ if we want to maintain integrity anyway, say for something simple like a one-to-may relationship?

Erlang User of the Year 2010

At the Erlang User Conference in Stockholm, I received the thirteenth "Erlang User of the Year" award, and was presented with the traditional gift, a beautiful hand-carved Erlang logo made in wood by Kanevad in Gamla Linköping.

I have to admit that I did not even know this award existed, but I am honored to join the ranks of Chris Piro and Eugene Letuchy (Facebook chat, 2009), Damien Katz (CouchDB, 2008), John Hughes (QuickCheck, 2007), Alexey Shchepin (ejabberd, 2006), The HiPE team members 2005, Michaël Rémond (ejabberd, 2004).

The award is given by the Erlang community for technical contribution; in this case my work on Erjang - A JVM-based Erlang VM. I was quite surprised (pleasantly of course) by this since I am by no stretch an Erlang old-timer. I wrote my first line of Erlang code a little over a year ago during JAOO 2009 when Francesco Cesarini was giving an Erlang tutorial.

The Erjang project is doing really well. For many systems, Erjang is roughly twice as fast as Open Source Erlang (BEAM). Recently, we've been working on completing more of the core functionality, and we're already at a level where we can boot and run complex systems like Riak and RabbitMQ.

Since I did not really have time to prepare an acceptance speech, I'm sorry I did not get around to thanking all the people that have been supporting this project. Most importantly I should thank Erik Søe Sørensen, who was the first guy to stand up and contribute code to this lunatic project, my family (for living with my lunacy on a daily basis for a year now), and also the entire Erlang community for being surprisingly helpful and positive to this (literally) bastard project.

Once again, Thank you all!

Kresten Krab Thorup

Erjang - Interpreted Mode

Erik has been hacking the interpreter the last few weeks, and now we're finally at a state where it is useful (though there are still some issues).

Interpreter??

Yep, now Erjang has two modes of operations.

• The default is the Eager JIT mode, meaning that modules are compiled from .beam to .jar-with-class-files eagerly as they are encountered as OTP boots up. This results in relatively slower start-up time, but fast execution.
• The new option ej +i lets you run Erjang with the byte-code interpreter execution mode; which starts up faster, uses less memory, but runs slower.

One effect of this can be seen from the launch time of Erjang. Look at this

prompt$ rm .erj/*   ## delete class cache
prompt$ echo 'erlang:halt().' | time ej
Eshell V5.8  (abort with ^G)
1>        13.96 real        14.45 user         1.86 sys

Wow, that took 14 seconds to launch the Erjang REPL. Now, doing the same thing again should speed things up because now the .beam files have been compiled; see here:

prompt$ echo 'erlang:halt().' | time ej
Eshell V5.8  (abort with ^G)
1>         4.67 real         4.61 user         0.29 sys

Wow! the second launch was much faster because the Eager JIT does not have to pay the price of the up-front compilation.

So, running the interpreted Erjang (notice the +i there) gives you:

prompt$ echo 'erlang:halt().' | time ej +i
Eshell V5.8  (abort with ^G)
1>         2.47 real         2.64 user         0.20 sys

Wow, its only ~55% of the execution time.

You might be puzzled why this is faster than the pre-compiled case above. The reason is that Java byte code loading is fairly slow, and that case ends up loading a lot of java code. And the Java JIT has not yet had a chance to optimize the code.

Unfortunately, running BEAM (the erl command) yields...

prompt$ echo 'erlang:halt().' | time ej +i
Eshell V5.8  (abort with ^G)
1>         0.24 real         0.11 user         0.02 sys

10 times faster! Urgh. Fortunately for most erlang systems, launch time is not significant... but still. Over time (when the Java JIT kicks in) the performance of Erjang and BEAM are currently at-par.

Memory Usage

Having the interpreter provides two new interesting possibilities. First, it brings Erjang within reach of running on an Android machine, because (a) this execution mode does not involve runtime generation of code, and (b) the memory load is much smaller. Lets try to launch mnesia and inspect memory usage. see:

First, look at an interpreted mode...

prompt$ ej +i
1> mnesia:start().
ok
2> 'java.lang.System':gc().
ok
3> erlang:system_info(allocated_areas).
[{'non_heap:CMS Perm Gen',134217728,10628584},
 {'heap:CMS Old Gen',50331648,7618216},
 {'heap:Par Survivor Space',1638400,0},
 {'heap:Par Eden Space',13500416,1790200},
 {'non_heap:Code Cache',2560000,2529664}]

Next, compiled mode...

prompt$ ej
Eshell V5.8  (abort with ^G)
1> mnesia:start().
ok
2> 'java.lang.System':gc().
ok
3> erlang:system_info(allocated_areas).
[{'non_heap:CMS Perm Gen',134217728,35051856},
 {'heap:CMS Old Gen',50331648,13346696},
 {'heap:Par Survivor Space',1638400,0},
 {'heap:Par Eden Space',13500416,416592},
 {'non_heap:Code Cache',2494464,2483520}]

To make it more readable, here is a table ...

	compiled	interpreted
Old Gen	13,346,696	7,618,216
Survivor Space	0	0
Eden Space	416,592	1,790,200
Perm Space (Java byte code)	35,051,856	10,628,584
Code Cache (native code)	2,483,520	2,529,664

With the compiled mode, memory adds up to ~55MB, in interpreted mode, ~20MB. Over time when the Java JIT kicks in, this difference is going to be even more evident.

Another interesting measure, is the memory usage before and after starting mnesia. With ej +i starting mnesia adds ~1MB total memory usage (similar to BEAM); but with compiled mode, starting mnesia adds ~9MB memory usage. So, while the baseline for a booted OTP is higher with Erjang +i than BEAM (~19MB vs ~3MB), the incremental memory usage when loading code and running is the same order of magnitude.

HotSpot Erjang

Another interesting potential would be to combine the Erjang interpreter an the Erjang JIT, so that code starts up interpreted and then if something is used "enough" it will get JIT'ed to java byte code. This is similar to how HotSpot Java does it (but is also one of the things that Oracle has patented and was recently suing Google over ...).

In such a scenario, there are some more interesting optimizations that would make sense; because if we have more information then it might well make sense to do more agressive optimizations and trade off some extra byte code generated for more speed.

Getting there...

As I hinted above, there are still issues with the interpreted mode; one is that interpreted mode uses more stack stack space; another is that there are some still some bugs lurking in there. At lease when running rabbitmq, something doesn't boot correctly. So, there is still some wat to go.

GOTO Top 20 Talks

Here's a quick tally of the top talks at last week's GOTO Aarhus conference. The rating is based on the red, green and yellow "exit-poll", computed as (#green - #red) / #total votes.

Make sure to evaluate the conference here and join the raffle for one of five signed copies of Martin Fowler's new book Domain Specific Languages.

Mario's technical talk on ASP.NET MVC went straight to the top. Two people made it onto the list twice: Tim Bray and Brian Goetz. Gregor Hohpe who stepped in for a cancellation with a talk on Near Field Communication made #2.

1. Mario Szpuszta: Fall in Love with ASP.NET MVC - Testable, Dynamic Web Applications with .NET

2. Gregor Hohpe: NFC - Not For Children?

3. John Hughes: Testing Asynchronous Behaviour

4. John Maloney: Scratch: Making Programming Easy and Fun

5. Dan North: From months to minutes - upping the stakes

6. Michael T. Nygard: Dev and Ops Cooperation when the Worst Happens

7. Tim Bray: Doing It Wrong

8. Kati Vilkki: Becoming Agile is HARD for a Large Organization

9. Mary Poppendieck: What is this thing called "Pull"?

10. Tim Bray: The Mobile Imperative

11. Brian Goetz: Java Future at Oracle

12. Jim Webber: Lessons Learned in Large HTTP-Centric Systems

13. Stuart Halloway: Clojure Protocols are not Interfaces!

14. Jez Humble: Continuous Delivery

15. Jonas Jacobi: HTML5 WebSockets

16. Brian Goetz: Data parallelism in Java

17. Bill Pugh: Defective Java: Mistakes that matter

18. Arjen Poutsma: Having fun with the RestTemplate

19. John Allspaw: Dev and Ops Cooperation at Flickr

20. Dan Ingalls: Forty Years of Fun with Computers

Mind you, being #20 out of 84 talks is pretty good, and represents 83% approval.

Once again, we have talks from the so-called "vendor track" pop into the top talks; this. Jez Humble's and Arjen Poutsma's talks were very well received.

Personally, I'm sorry that I only made it to 5 of the talks on this list. With so much to choose from (and my various other obligations) it was hard to make it.

Over all distribution

The over all distribution of evaluations look like this, and I'm happy that 80 out of 84 talks total are above expectations. Great kudos to all the speakers and the program committee for making this happen!

As can be seen, four talks fell below expectations; and while that's obviously too many it gives us ideas for what to improve for next year's GOTO Aarhus. I think those bottom ones are one of two reasons: 1. misalignment of expectations, or 2. some speakers found it very difficult to present in the large room where it was difficult to interact with the audience.

And please don't ask who's at the bottom. I won't tell.

RabbitMQ on Erjang

It's been quiet on Erjang for a while, and I've been busy with other things. Going to JavaZone two weeks ago however, got me all psyched up again on my little Erlang on the JVM project.

I met some guys from VMWare/SpringSource and we talked about RabbitMQ, so I spent some late nights these past weeks getting things working. The prospects of this? How about a SpringSource developer wanting to use RabbitMQ? Eventually, this would be a great option as an integrated part of a pure Java development environment for say, Eclipse.

It's at the point now where I can run a good deal of the unit tests, though many still fail. You can read more about the progress below.

The major improvements are

• Mnesia is much more stable now
• TCP/IP error handling is much improved

It's great to work specifically on something like RabbitMQ because it already has a large base of existing test cases. Further, RabbitMQ is plain erlang i.e., no NIFs or linked-in drivers, so it is a perfect candidate for Erjang.

Looking forward to demo/talk about this next week at JAOO. Feel free to follow Erjang at GitHub if you want to stay in the loop on this.

Additions/Fixes...

To get to this point, I added a number of new BIFs, and fixed various bugs. You can see the glory details in the commit log here, but here is an outline of the most important fixes/additions:

• Implement erlang:hibernate which is used in gen_server2 and a lot of the other infrastructure modules that come with RabbitMQ. Hibernating is implemented as an exception which is caught at the top-level of the process so as to remove the call stack before going to sleep.
• Fix a couple of issues with the tcp_inet driver. Connections shutdown procedures didn't work completely, and rabbit depends on the owner dying to close connections. Further, rabbit uses the tcp option {exit_on_close, false} to allow outgoing communication to flush before the actual tcp socket is closed, and Erjang had some issues with that too.
• Improve the driver select mechanism.
• Implement a few more options for the efile driver. Rabbit uses the FILE_SETOPT driver option to configure read-ahead strategies for file usage, and also depends on the integer (file descriptor) returned by FILE_OPEN.
• In ETS, I had to implement a few more bif used specifically by Rabbit/Mnesia, such as ets:repair_continuation, ets:update_counter/3 (the 4-tuple variant).
• Various fixes to the BEAM -> JVM compiler that I stumbled upon when running Rabbit. There were some cases where the type inference was slightly off, and I optimized the cases when erlc uses {x,1023} as a scratch register.

I still have an issue with some of the functions in io_lib being non-tail recursive, causing stack overruns. The problem is that this often happens when Erjang tries to print an error message, and so it doesn't print the reason why it dies. And in some cases, the error_logger process itself dies from a stack overrun, bringing down the entire VM.

Going forward, it looks like the issues yet-to-be-fixed are either simple bugs like variants of arguments that are not supported by my BIF implementations, and (more tricky) some cleanup issues with processes.

Stay tuned! I'll talk more about these things at JAOO next week, QCon/SF and ErlangFactory Lite LA in November, YOW! Australia in December. If you want me to come and speak please let me know.

Booting RabbitMQ on Erjang

Booting rabbit right now happens with the following command line

ej -boot /var/lib/rabbitmq/mnesia/rabbit@renaissance/plugins-scratch/rabbit \
   -pa ./ebin -name rabbit2@renaissance.local \
   -sasl sasl_error_logger '{file,"sasl.log"}'

The bootup sequence looks like this (running the latest rabbitmq 0.9.1 build from hg) on Erjang/OTP R14.

=INFO REPORT==== 29-Sep-2010::12:03:51 ===
    alarm_handler: {set,{{disk_almost_full,"/"},[]}}

=INFO REPORT==== 29-Sep-2010::12:03:51 ===
    alarm_handler: {set,{{disk_almost_full,"/Volumes/StockMeter"},[]}}

=INFO REPORT==== 29-Sep-2010::12:03:51 ===
    alarm_handler: {set,{{disk_almost_full,"/Volumes/Flash"},[]}}
|
=INFO REPORT==== 29-Sep-2010::12:03:51 ===
    alarm_handler: {set,{system_memory_high_watermark,[]}}
+---+   +---+
|   |   |   |
|   |   |   |
|   |   |   |
|   +---+   +-------+
|                   |
| RabbitMQ  +---+   |
|           |   |   |
| v%%VSN%%  +---+   |
|                   |
+-------------------+
AMQP 0-9-1 / 0-9 / 0-8
Copyright (C) 2007-2010 LShift Ltd., Cohesive Financial Technologies LLC., and Rabbit Technologies Ltd.
Licensed under the MPL.  See http://www.rabbitmq.com/

node           : rabbit2@renaissance.local
app descriptor : /Users/krab/Projects/rabbitmq-server/ebin/rabbit.app
home dir       : /Users/krab
cookie hash    : pIGTBl0W0HD2xiNguwnLoQ==
log            : tty
sasl log       : sasl.log
database dir   : /Users/krab/Projects/rabbitmq-server/Mnesia.rabbit2@renaissance.local
erlang version : 5.8

starting file handle cache server                                     ...
=INFO REPORT==== 29-Sep-2010::12:04:05 ===
Limiting to approx 10140 file handles (9124 sockets)
done
starting worker pool                                                  ...done
starting codec correctness check                                      ...done
starting database                                                     ...found mnesia running id=<2.0.592>
Sep 29, 2010 12:04:05 PM erjang.EModuleManager$FunctionInfo$1 invoke
INFO: MISSING mnesia_sup:prep_stop/1

=INFO REPORT==== 29-Sep-2010::12:04:05 ===
    application: mnesia
    exited: stopped
    type: permanent

=INFO REPORT==== 29-Sep-2010::12:04:06 ===
Process <2.0.2031> exited abnormally without links/monitors
exit reason was: shutdown
erjang.ErlangExit: shutdown
    at erjang.m.erlang.ErlProc.exit(ErlProc.java:468)
    at erjang.m.mnesia_bup.mnesia_bup.install_fallback_master__2(mnesia_bup.S:207)
    at erjang.m.mnesia_bup.mnesia_bup$FN_install_fallback_master__2.go(Unknown Source)
    at erjang.EProc.execute(EProc.java:498)
    at kilim.Task._runExecute(Task.java:400)
    at kilim.WorkerThread.run(WorkerThread.java:47)

done
-- external infrastructure ready
starting logging server                                               ...done
starting exchange type registry                                       ...done
starting exchange type direct                                         ...done
starting exchange type headers                                        ...done
starting exchange type topic                                          ...done
starting exchange type fanout                                         ...done
starting statistics event manager                                     ...done
-- kernel ready
starting guid generator                                               ...done
starting cluster delegate                                             ...done
starting alarm handler                                                ...done
starting memory monitor                                               ...done
starting node monitor                                                 ...done
-- core initialized
starting empty DB check                                               ...
=INFO REPORT==== 29-Sep-2010::12:04:07 ===
Memory limit set to 1535MB.
done
starting queue supervisor and queue recovery                          ...
=INFO REPORT==== 29-Sep-2010::12:04:07 ===
Added vhost <<"/">>

=INFO REPORT==== 29-Sep-2010::12:04:07 ===
Created user <<"guest">>

=INFO REPORT==== 29-Sep-2010::12:04:07 ===
Set user admin flag for user <<"guest">> to true

=INFO REPORT==== 29-Sep-2010::12:04:07 ===
msg_store_transient: using rabbit_msg_store_ets_index to provide index

=INFO REPORT==== 29-Sep-2010::12:04:07 ===
msg_store_persistent: using rabbit_msg_store_ets_index to provide index

=ERROR REPORT==== 29-Sep-2010::12:04:07 ===
msg_store_persistent: rebuilding indices from scratch
done
starting exchange recovery                                            ...done
-- message delivery logic ready
starting error log relay                                              ...done
starting networking                                                   ...
=INFO REPORT==== 29-Sep-2010::12:04:07 ===
started TCP Listener on 0.0.0.0:5672
done
-- network listeners available

broker running
Eshell V5.8  (abort with ^G)
(rabbit2@renaissance.local)1> 

Running the tests from rabbitmq-java-client

And we can now run some unit tests. I've been poking around in the rabbit-java-client unit tests. Here are some screen shots...

The main tests for rabbitmq-java-client gets most of the way there, but currently dies with a bug in Erjang's ETS implementation when running the 500 producers/consumer test. Well well, can be fixed.

/com/rabbitmq/examples/TestMain.class
com.rabbitmq.examples.TestMain : javac v50.0 on 1.6.0_20
Channel 0 fully open.
Got message (4 left in q): (0) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Got message (3 left in q): (1) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Got message (2 left in q): (2) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Got message (1 left in q): (3) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Got message (0 left in q): (4) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Drained, remaining in batch = 0.
com.rabbitmq.examples.TestMain$BatchedTracingConsumer@6db3f829.handleConsumeOk(amq.ctag-jM3S9gxr4aSTUqoQC71+lA==)
com.rabbitmq.examples.TestMain$BatchedTracingConsumer@42698403.handleConsumeOk(amq.ctag-lL8JbFN2bq2oWxUgeDTuFQ==)
Async message (0,noack): (5) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Async message (0,ack): (6) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Async message (1,noack): (7) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Async message (1,ack): (8) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Async message (2,noack): (9) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Async message (2,ack): (10) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Async message (3,noack): (11) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Async message (3,ack): (12) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Async message (4,noack): (13) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Async message (4,ack): (14) On the third tone, the time will be Wed Sep 29 12:13:33 CEST 2010
Acking batch.
com.rabbitmq.examples.TestMain$BatchedTracingConsumer@6db3f829.handleCancelOk(amq.ctag-jM3S9gxr4aSTUqoQC71+lA==)
com.rabbitmq.examples.TestMain$BatchedTracingConsumer@42698403.handleCancelOk(amq.ctag-lL8JbFN2bq2oWxUgeDTuFQ==)
About to publish to topic queues
About to drain q1
Got message (2 left in q): B
Got message (1 left in q): C
Got message (0 left in q): D
Drained, remaining in batch = 7.
About to drain q2
Got message (0 left in q): C
Drained, remaining in batch = 9.
About to drain q3
Got message (1 left in q): C
Got message (0 left in q): D
Drained, remaining in batch = 8.
About to try mandatory/immediate publications
Handling return with body one
Returned: #method(reply-code=312,reply-text=NO_ROUTE,exchange=mandatoryTestExchange,routing-key=)
 - props: #contentHeader(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
 - body: one
Handling return with body two
Returned: #method(reply-code=312,reply-text=NO_ROUTE,exchange=mandatoryTestExchange,routing-key=)
 - props: #contentHeader(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
 - body: two
Handling return with body three
Returned: #method(reply-code=313,reply-text=NO_CONSUMERS,exchange=mandatoryTestExchange,routing-key=)
 - props: #contentHeader(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
 - body: three
Handling return with body four
Returned: #method(reply-code=313,reply-text=NO_CONSUMERS,exchange=mandatoryTestExchange,routing-key=)
 - props: #contentHeader(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
 - body: four
Got message (0 left in q): five
Drained, remaining in batch = 0.
Completed basic.return testing.
Got message (4 left in q): (15) On the third tone, the time will be Wed Sep 29 12:13:34 CEST 2010
Got message (3 left in q): (16) On the third tone, the time will be Wed Sep 29 12:13:34 CEST 2010
Got message (2 left in q): (17) On the third tone, the time will be Wed Sep 29 12:13:34 CEST 2010
Got message (1 left in q): (18) On the third tone, the time will be Wed Sep 29 12:13:34 CEST 2010
Got message (0 left in q): (19) On the third tone, the time will be Wed Sep 29 12:13:34 CEST 2010
Drained, remaining in batch = 0.
Handling return with body mandatory
Returned: #method(reply-code=312,reply-text=NO_ROUTE,exchange=,routing-key=bogus)
 - props: #contentHeader(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
 - body: mandatory
Handling return with body immediate
Returned: #method(reply-code=313,reply-text=NO_CONSUMERS,exchange=,routing-key=bogus)
 - props: #contentHeader(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
 - body: immediate
Got message (1 left in q): normal
Got message (0 left in q): mandatory
Drained, remaining in batch = 8.
Closing.
Leaving TestMain.run().
WON'T write statistics.
WILL use server-side auto-acking.

When process are cheap, who needs timeouts?

I Hate Time-Outs

A recurring issue in the design of concurrent and/or distributed systems is to choose those dreaded time out values. It seems that they are inevitable, doesn't it? They seem arbitrary, baked-in in an way that often make them hard to change and mange. In my experience, they are always wrong somehow, or at least need a lot of trial-and-error hand tuning to get right.

Imagine if we could do away with them all together. Woudn't that be a relief?

The other day when I was explaining some Erlang issues to Steven Freeman (@sf105), he summed up my explanation with those magic words: "When process are cheap, who needs timeouts?" And it's so true.

This ideas I was describing to him came from a recent discussion on the erlang-questions mailing list. That discussion started with a question about the default timeout in erlang's standard RPC mechanism gen:call; which used to do an rpc invocation on a generic server. It has a standard timeout of 5 secs, so when one process performs an RPC request to another process inside the erlang ecosystem, it has a default timeout of five seconds. Why? In the end someone authoritatively explaining that it was really a mistake in the original design to have a default timeout different from infinity, because it's not really needed.

The reason why timeouts are not needed is that when you use the gen:call mechanism the caller monitors the callee while waiting for a reply. That means the caller knows if the target of the rpc dies. Why have a timeout then?

Timeouts really belong at the edges of the system, not inside it. The relevant timeout is one that the ultimate caller -- the client -- should be concerned with. And in stead of specifying a timeout, why not support some kind of cancel mechanism?

When interacting with some external resource, you can feasibly use a timeout to decide that the resource has become unavailable. (This strategy is also part of what erlang nodes use to determine liveliness of connected nodes, in combination with tcp sockets being broken.)

Timeouts waiting for some response to arrive should only be used at the edges.

Cancelling requests

Another case of timeout at the "edges of the system" is a client sending a request to your server. Think of a HTTP request which is cancelled by the user by closing the connection; or it could be an explicit cancel request in some other protocol. When such a cancellation arrives, you'd want to be able to propagate this cancellation to the part of your program which is working to carry out the request. In this case, the normal timeouts doesn't help you either.

Luckily, erlang is great for something like this. A standard pattern is to create a process for each request, and then ... have processes that are spawned by the "request process" monitor or link to the original request process. That allows you to propagate the cancel request the other way -- down the RPC call chain. I think of it as an exception propagating in the opposite direction. It propagates from the caller to the callees. Awesome!

I have not been able to find code in OTP that covers this usage model, and even though this code probably doesn't do exactly the right thing

 ...
 handle_request({do_something, Args}, {CallerPID,_}=ReplyTo, State) ->
    JobRunner = spawn(fun() ->
       link(CallerPID),
       {ok, Result} = compute_do_something( Args, State )
       unlink(CallerPID),
       gen:reply(ReplyTo, Result)

    end),
    {noreply, State}.

i.e., while executing a long-running job compute_do_something, link with the requestor so that the JobRunner gets killed if the requestor dies.

Ideally, you'd want to use the less intrusive monitor option rather than linking, but that requires more integration across modules. But I like the idea of an "exception propagating in the opposite direction" to describe a cancel request.

Erlang Message Queues and RPC

I've been hacking Erlang for close to a year now, and in the process of learning Erlang there as been quite a few surprises. Here I'd like to share some of these thoughts, and this is the first of a series of postings I intend to do about this.

In Erlang, each process has only one! message queue

This boggled me for a while. If each process/actor has only one message queue, then how can it wait for multiple things in an efficient manner?

In my programming career, I've done many different things that involve concurrency or event-driven programming, and usually it boils down to some kind of event and/or select loop in which you usually want to wait for one of several things to happen. Not just one. With Erlang a process has only one mailbox, so how can you wait for more than one?

The answer is that in Erlang, you essentially multiplex multiple conversations on the same message channel (i.e. the same mailbox). The key is that you include a "conversation id" in messages pertaining to a given interaction, and then match on multiple such messages using so-called selective receive. When you use selective receive, a process can wait for one of several message patterns to arrive. The patterns will typically be tuples containing the conversation id.

receive
   {ID1, Value1...}  -> ... do something in conversation 1... ;
   {ID2, Value2...}  -> ... do something in conversation 2 ... ;
   after 100 -> timeout after 0.1s
end

Executing the receive operation will first scan all existing messages in the current process' mailbox and see if any one of those messages match either tuples that containID1 or ID2 in the first position; and then it will sleep (for up to 100ms) until more messages arrive and then try the matching on those new messages.

Having selective receive is extremely useful and important for managing multiple concurrent conversations in your system; as e.g. Ulf Wiger explains expertly in this presentation from QCon London 2010.

The server + rpc problem

One issue with this however, is that scanning all messages does not scale well if the message queue is long, and you're looking for a specific message. And that can often be the case when doing an "rpc-like" message exchange from the context of a "server-like" process that has many incoming requests.

Take this example of an rpc-like call to the process OtherPID, which is done by sending a message and receiving a reply; something like this:

send_receive_rpc(OtherPID, RPCMessage) ->

  % create unique "conversation id"
  Ref = create_ref(),

  % send message to OtherPID, including where to send the response
  % and the "conversation id" to include in the response.
  OtherPID ! {Ref, self(), RPCMessage},

  % now, wait for a tuple {ID, Result}; that corresponds to the call
  receive 
    {Ref, RPCResult} ->
        RPCResult
  end.


rpc_server_loop(State) ->
   receive
      {Ref, Sender, RPCMessage} ->
          {RPCResult, NewState} = do_work(Message,State),
          Sender ! {Ref, RPCResult},
          rpc_server_loop(NewState)
   end.

In the above code, we create a unique value Ref that can be used to match the specific response to this message, and then we use that to match the reply. OK, that's elegant, but is it efficient?

Upon entering the receive loop in this case waiting for the RPC response, we need to first scan all existing messages in the process' mailbox and see if any one of those messages match {Ref, _} (a lost cause since the Ref is unique so there are none such) and then we go to sleep and wait for the reply.

So ... if the rpc client is a process that itself has a long message queue then this is fundamentally flawed. ... How on earth did Erlang ever succeed to get this far with this problem?

Luckily it has been "fixed" by a kind of peep-hole optimization in the Erlang compiler that recognizes receive statements that do not need to scan the existing message queue. The fix is in Erlang R14, here. The optimization works by a simple flow analysis that can (in some cases) determine that the receive does not need to match messages that were already in the message queue by the time the erlang:create_ref/0 was called. As a result, the code is encoded into something like this pseudo-code; recv_mark saves a pointer to the end of the message queue, and recv_set validates some preconditions and instructs the subsequent receive to start scanning from the save point.

recv_mark(UID),
Ref = create_ref(),
OtherPID ! {Ref, self(), RPCMessage},
recv_set(UID),
receive 
  {Ref, RPCResult} ->
      RPCResult
end.

This is a good reason to use gen_server:call which does something like the above for you rather than hand-crafting your own code to do message sends ... because that function exhibit the exact code pattern that ensures that the peep-hole optimization kicks in.

And you have to be careful; it only works for create_ref/0 and monitor/2; and only when the created ref occurs in all clauses of a subsequent receive expression. For rpc:call for instance, it doesn't work because it uses erlang:spawn_monitor not one of the two functions mentioned above. [It could be made to work though, but that needs another fix that can cover the flow analysis for the ref in the {Pid,Ref} tuple returned by spawn_monitor.]

One obvious "real fix" for this would be to have multiple message queues (or channels) as Go has it plus selective receive; and we could easily create a new unique queue for the purpose of the RPC.

ChannelRef = create_channel(),
OtherPID ! {ChannelRef, RPCMessage}
from ChannelRef receive 
    Reply -> Reply;
from OtherChannel receive 
    ... -> other action
end

Not an "easy fix", because it would fundamentally change the language; but seems like a better choice should one choose to design a new language in this space. And it is sort of amazing that Erlang gets away with being so simple!; it's just kind of confusing (to me at least) the way incoming messages are mixed up with RPC replies in the same mailbox.

Another "fix" that might make sense would be to introduce a new language construct to do explicitly what the compiler does implicitly; sort of a stack-oriented way of pushing a new (empty) mailbox onto an implicit mbox queue.

push_mbox
  ... send rpc request ...
receive
  ... match rpc reply ...
end

Which would then make the receive clause disregard messages that are already in the mailbox at the time push_mbox was called. The completion of the receive clause would then need to merge any new messages into the end of the previous mbox. this language construct would be fairly easy to add, because the instructions to handle it are already in the VM.

Stray Messages

Another "issue" with using selective receive is what happens if messages arrive that you do not understand. Erlang folklore says that you should have somewhere in your main loop, where you simply discard such unknown messages (and perhaps log them as errors).

But in the presence of multiple concurrent "coversations" multiplexed onto a single message queue, it can be very difficult to know when we have a safe point where only certain kinds of messages make sense.

In this context it would also make sense to use a model with multiple channels, because then it would at least be easier to establish a contract (kind of a protocol/interface for meaningful sequences of messages occurring on a channel).

Scala's actors - even though they have channels - do not seem to support this pattern of concurrently doing selective receive across a number of channels. That would be very cool.

Other RPC-goodies

Digging around the code you also come realize how many cool things there are in the OTP frameworks that you don't think of. One such thing is how doing an RPC will actually make you monitor the callee. The code in there is something like this Warning pseudocode!:

do_rpc_call(OtherPID, RPCMessage) ->
  MonitorRef = erlang:monitor(process, OtherPID),
  OtherPID ! {MonitorRef, self(), RPCMessage},
  receive 
    {'DOWN', MonitorRef, _, _, Info} -> 
        {badrpc, ...};
    {MonitorRef, Result} ->
        erlang:demonitor(MonitorRef),
        {ok, Result}
  end.

The MonitorRef doubles as the "conversation id". What this code does is that while you're doing an RPC call, you also monitor the callee; so if the OtherPID dies, then you get a sensible reply. If OtherPID is on a remote node, then erlang distribution's keep-alive messages may eventually deem the remote end dead and thus deliver appropriate local DOWN messages. Way cool.

You could also do this the other way around; have the callee monitor the caller while it does some computation; canceling the an expensive computation if the caller goes away. That would indeed make sense if the caller was an HTTP connection handler for an impatient and-or slow user!

Who's waiting for who?

Another - related issue - with Erlang's mailbox mechanism is that it is impossible to know who is waiting for who. So even though you encode the dependencies between processes quite explicitly; there is no way to know which process is responsible for producing the reply you're waiting for.

At the recent Java Language Summit Doug Lea gave an excellent talk, in which one of the points was that to schedule light-weight concurrency effectively it is very valuable to have a dependency graph in the scheduler so that you know which processes are waiting for which other processes. The point is that it reduces "global work" to spend processing time finishing work that has already been started before you take on (or generate) more work.

In Erlang, the dependencies between the RPC client and the RPC server is entirely encoded in some "unstructured" data that flows on the message channels, and there is no easy way for the scheduler to exploit this information.

Another potential value of such information would be that you can (At least partially) prevent priority inversion, in the case where a high-priority rpc client sends a request to a lower priority rpc-server.

Perhaps some of that information could be captured the same way the current "fix" is in Erlang/OTP, by fixing the case for the common framework code; and perhaps that would be enough to capture such dependencies for most cases. I don't know.

Conclusion?

For writing systems that handle multiple concurrent conversations with other agents/processes/services, Erlang's model is IMHO the only thing out there that works. If you disagree, you should see Ulf Wiger's QCon presentation Death by Accidental COmplexity one more time.

I believe systems that have these patterns are quite common.

But I still think we could do even better.

References

If you google "selective receive" you can find numerous references to these issues. Several different fixes has been suggested, and the problem seems to come up every now and then.

See for instance this post from the RabbitMQ community, or this one. Seems that the usual way to handle this is to put a proxy process in front, which buffers the messages so that not too many messages end up in the "controller".