Common Lisp bindings now part of ZeroMQ 2.0

Today Vitaly Mayatskikh announced the inclusion of his Common Lisp ZeroMQ bindings into the ZeroMQ 2.0 source tree. ZeroMQ 2.0 is a high-performance messaging system, which is great news if you are trying to build distributed systems in Common Lisp.

Previously, available Free Software alternatives for Common Lisp messaging included CL-XMPP (XMPP client only) and CL-RABBIT (Lispworks only).

Note that ZeroMQ 2.0, unlike version 1, no longer implements AMQP, due to performance reasons.

The impossibility of global state

(This post is a write-up of some ideas about concurrency that I've briefly discussed before).

By far the bulk of concurrency practice is aimed at dealing with systems based around shared, mutable state. Participants are asked to imagine a world with a global state whose changes are instantaneously observable by everyone. This world is a physical impossibility.

The global state illusion is based on elaborate consensus mechanisms. The concept only works because our perception of the flow of time is fixed vis-a-vis the rest of the world, which happens to be just quick enough to execute consensus algorithms over large networks that we don't get annoyed.

The common refrain to criticisms of the global state illusion is "but think of the bankers!" The consistent bank account with atomic deposits and withdrawals is an example found in almost all textbooks on concurrency. It is a neat and self-contained pedagogical tool for discussing the mechanisms needed to construct a global state illusion, but unfortunately its pervasive use seems to have done immense damage to any attempts to actually reason about concurrent systems.

Networked credit banking existed in 4th century BC Egypt. Banks still accept checks. Your bank account still has overdraft fees. Bank accounts work by reifying transactions as first-class objects. That this idea is considered novel is a testament to the pervasive confusion surrounding concurrency today.

What is the global state illusion really needed for? Distributed shared simulations. Most commonly seen in the guise of videogames. There, all the participants really do need to see a consistent view of the world with changes occurring in quantized time steps.

The key insight behind the global state illusion is that it is not time-scale independent. Whether the illusion can be maintained depends on how the participants perceive the flow of time relative to how quickly the mechanisms behind the illusion work. Banks could have implemented a pen-and-paper two-phase commit protocol for deposits and withdrawals before the advent of telecommunications. It would not have been very useful. This is the same exact issue affecting cache-coherency protocols in multi-core processors, at the micrometer scale.

The way forward in concurrency practice is the abandonment of the global state illusion in favor of constructing systems based on how concurrency works in the real world, taking cues and patterns from organizational and biological systems. This will entail the use of techniques from functional programming and linear logic.

Gelernter

Two years ago I posted a review of Nicholas Carriero and David Gelernter's How to write parallel programs. Today I came across an Edge roundtable discussion with David Gelernter (via Patrick Logan).

The entire discussion is fascinating, and I recommend reading up more on Gelernter's work if you are not familiar with it before watching the interview.

For me, there were three new insights from the roundtable:

• Message passing is the assembly language of distributed systems.


• Databases are a way to communicate through time. Networks are a way to communicate through space. Both are symmetrical. The point of Linda was to unify them.


• The most powerful computers today are botnets. Botnets are also on the leading edge of distributed systems techniques.

In the second video, Gelernter discusses Lifestreams. It's amazing how closely FriendFeed, Twitter, and the new Facebook are all following the roadmap established by Lifestreams over a decade ago.

Cassandra of Facebook, or A Tale of a Distributed Database, part 2

This is the third in my series of blog posts about Facebook's Cassandra distributed database. Part 0 discussed the background of the problem Cassandra is meant to address, while part 1 gave an overview of the distributed systems techniques employed by Cassandra. This post is a scattering of my observations from reading the code.

The most obvious feature of Cassandra's code is the pervasive presence of checksums in communication and data handling modules. Checksums are used both to detect data and message corruption, and as a means of finding out if replicated nodes are out of sync.

Cassandra is a production system with a large deployment and the management features present reflect this. There is code for measuring node statistics and sending them off to a collection server. Nodes register themselves with both Zookeeper and JMX cluster-management services, although exactly what both of those are used for at Facebook is unclear as their cluster-management tools have not been released as free software.

The mechanism for selecting nodes for replication can be customized to use metadata about machines to choose nodes on redundant networks/power grids/backplanes/etc to increase availability. The strategy provided in the code indicates that Facebook structures their Cassandra deployment using IPv4 addresses so that all the machines with the same third octet of an address are in the same rack, and all machines with the same second octet are in the same datacenter.

The gossip protocol, besides being used for failure detection, is also used for load-balancing the cluster. Workload is measured as the number of requests per number of keys in a given time window that are handled by a node responsible for a particular interval of the key ring, and is disseminated by each node to the others. To alleviate high load, a node may move itself around on the key ring to more evenly distribute the keys between itself and a lightly loaded neighbor. The approach is an implementation of Abdallah and Le's Scalable Range Query Processing for Large-Scale Distributed Database Applications.

The gossip protocol itself is rather interesting, featuring three stages (gossip message, ack, ack-ack).

An interesting artifact is the com.facebook.infrastructure.continuations package, which has some stubs based on the Javaflow bytecode continuation transformer. You would guess this would be used for the server code since it's based on non-blocking IO, but the server code is actually hand-coded staged event-driven. The stubs in the package aren't used anywhere else, which means that some other Facebook project must be using those. I wonder what for?

The code itself can be found at http://code.google.com/p/the-cassandra-project/, but seems to be in the process of being moved to a different repository. In the meantime, there is an active git repository at http://wiki.github.com/jbellis/cassandra-dev that features community-submitted bugfixes and improvements.

To end on a light note, it is refreshing to find that the authors of Cassandra give a nod to the LOLspeak code commenting guidelines:

* 3. Set one of teh messages to get teh data and teh rest to get teh digest
// 4. SendRR ( to all the nodes above )
// 5. Wait for a response from atleast X nodes where X <= N and teh data node
* 6. If the digest matches return teh data.

Cassandra of Facebook, or A Tale of a Distributed Database, part 1

This is the second part of my examination of the Cassandra distributed database. Part 0 described the problem that Cassandra is trying to address; this post will describe the distributed systems techniques that it utilizes to that end.

The basis for this post is a presentation describing Cassandra by Prashant Malik, Karthnik Ranganathan, and Avinash Lakshman given to the Facebook staff and available online: http://www.new.facebook.com/video/video.php?v=540974400803#/video/video.php?v=540974400803 (hosted on Facebook, an account is required for viewing). James Hamilton provides a summary of the presentation on his blog.

Underlying Cassandra is the technique of consistent hashing. Using consistent hashing, the key space can be modeled as a circle of the keys' hash values, and nodes assigned to points on that circle, so that a node is responsible for an interval of the circle of key hashes between itself and its next neighbor. Nodes leaving or joining only affect their previous neighbor, and lookup has a very low communication cost. Due to its great simplicity and robustness compared to other mechanisms, consistent hashing is probably the most significant development in distributed systems in the 1990s - virtually all current peer to peer systems are based on it.

Cassandra uses a gossip protocol for determining cluster membership and for failure detection (and for load-balancing, discussed in the next blog post). Failure detection in this context means the various failure detector mechanisms that can be used to achieve consensus in asynchronous distributed systems with unreliable nodes. Given unlimited time to finish a task, it is impossible to distinguish a dead node from one that is slow - failure detectors use timing assumptions and heartbeat messages between nodes to provide an answer (with some degree of uncertainty), something that would otherwise require a synchronous network between nodes. Cassandra implements a failure detector based on Hayashibara et alia's The φ Accrual Failure Detector; the presenter points out that some "ghetto" failure detectors with simpler mechanisms do not work very well in practice.

Each interval of the hash circle is replicated on a set number N of nodes. Cassandra provides a choice of replication scheme: optimistic replication, which gives eventual consistency, or stricter approaches for writes and reads: quorum write and sync on read. Optimistic replication allows writes even if all nodes that are responsible for storing the particular key are down: the node receiving the request logs it and later hands it to the appropriate nodes when they come back up. Under the quorum write scheme, a write is not successful until at least a specified x number of the N total replicas have performed the write. It would seem that if x=N, then we get both monotonic read and write consistency. Sync on read synchronizes all replicas on each read request, which gives monotonic write consistency.

Cassandra of Facebook, or A Tale of a Distributed Database, part 0

One of the most natural and popular tendencies in the application of distributed systems theory is the construction of distributed databases with the hopes of providing improved availability and scalability (in the number of requests processed per second) via data replication. While distributed queries and transactions are obviously needed for some kinds of databases, what underlies every database is the ability to read and write a sequence of bytes to some place denoted by another sequence of bytes. In database jargon this is commonly called a storage engine, while in a distributed system this can be seen as a type of distributed shared memory.

Any kind of database providing access to more than one client at a time must deal with concurrency issues and hence provide some kind of consistency model for data access and manipulation. This can make constructing distributed databases either easier or harder, as the underlying consistency model of a chosen distributed implementation can be exposed directly as that of the database, or several distributed mechanisms have to be combined in complicated ways to provide the properties mandated by the stated consistency model of the database.

Choosing a given consistency model involves large trade-offs in the performance, simplicity and ease of maintenance of the distributed database implementation. Ideally, all the nodes of the system would play the same role, and adding new nodes would incrementally improve the availability, capacity, and number of requests the system can handle. Because of the communication and synchronization overhead necessary for implementing stricter consistency models, the improvement in those aspects as a function of nodes in the system displays the quality of diminishing returns (and may in some cases provide negative ones).

Fortunately, many applications do not require strict consistency models of their database. The first to go is usually the requirement for multi-operation transactions with the ACID property. This provides the greatest improvement in potential scalability as both the locking and multi-version timestamping mechanisms for implementing transactions degrade in performance with increased contention between nodes.

The fad du jour in distributed databases is the distributed non-transactional non-relational database. It seems there is a new project implementing a distributed key-value, document-oriented, or "loose schema" database announced every couple of months.

Despite the hype, only a handful of these projects have large deployments. Even rarer is coming across someone who has worked on more than one such database. These facts alone make Facebook's Cassandra stand out. Cassandra was developed and is currently deployed inside Facebook to handle the user messaging feature. One of the developers of Cassandra, Avinash Lakshman, has previously worked on Amazon's internal Dynamo distributed database.

Fortunately for those of us who do not work on messaging at Facebook, the Cassandra source code has been released under the Apache 2.0 free software license. This is the first in a trilogy of blog posts examining Cassandra's use of modern distributed systems techniques and my observations from sitting down and reading the code.

Peer-to-peer synchronized simulations

It seems it's been a while since I wrote anything about distributed systems. The next few posts should hopefully make up for that.

To kick things off with some light observations, I've recently (thanks to a whole bunch of people in my del.icio.us network) come across this 2001 article describing the multiplayer system of Age of Empires.

The system is a peer-to-peer simulation where input from each player in the form of commands is synchronized across all nodes in discrete "game turn" steps, whose real-time duration is calculated as a function of latency between the nodes and the framerate at which the nodes can render the simulation.

It's interesting to compare the Age of Empires system to that of Croquet (I've blogged about Croquet before). Like Age of Empires, Croquet is a peer-to-peer system. The TeaTime architecture behind Croquet can be thought of as a distributed object simulation where messages between objects are what gets synchronized across nodes.

Where Croquet and the AoE multiplayer system differ is in how time and synchronization is handled. TeaTime extends Reed's multi-version timestamping with the notion of pseudo-real time and uses that notion to enable deadline-based scheduling for message processing, with synchronization being achieved via two-phase commit. This allows the decoupling of simulation rendering speed from message synchronization. Unlike the AoE system, this approach is unaccomodating to high-latency nodes.

One thing I found interesing about the AoE article is the large number of synchronization bugs (a lot of them relating to PRNG generation) that the developers encountered.

Memory doubts

Patrick Logan recently posted a note expressing misgivings about the promise of transactional memory:

I can see someone making the argument in some domain I don't usually work in that shared memory threads are better than shared nothing message passing for performance reasons. Some hard-real time scenario (I think I was real tired when I wrote that. Must have been thinking back to the days when I'd get into "Is automatic garbage collection practical?" debates. Nobody has those anymore, do they? Because I've got some real good arguments for GC.), some huge number crunching scenario, etc. where every byte and every cycle has to count in the extreme. But the irony then is that STM seems far more suited to a language like Haskell, which is also unlikely to be suited for these performance scenarios.

My only fear is that for the masses including myself, we need *simple* mechanisms, and the fewer of those, the better. Shared nothing messages seem to scale up, out, and down. STM seems more complicated, and an incomplete solution, and only a solution for shared memory. No thanks, at least until bigger brains than my own can show me why I should change my mind. They seem sidetracked on this one.

With regards to transactional memory, the "performance argument" doesn't even need to be brought up - the second biggest objection to transactional memory (the first being that it is different than what's currently being used), is that even in the best case of no contention, there is a large overhead penalty imposed versus code using manual synchronization. Of course in the worst case you would have to throw out a whole bunch of transactions' worth of work and start again.

When it comes to scalability, transactional memory is just as limited as any other kind of transactional system. Published benchmarks already show a performance plateau on some of the larger Sun servers, and those have less cores and probably a much lower CPU-speed to memory latency (and bandwidth) ratio than any kind of upcoming 80-core Intel chip.

If anything, out of all available approaches message-passing is by far the fastest that can be implemented. Cache-coherency protocols introduce delays and seriously limit scaling in the number of processors in an SMP system, and even atomic instructions are so expensive that I've heard the overhead means that some of the lock-free datastructures currently being worked on don't perform well compared to the lock-based ones they are supposed to be replacing.

However, I would not by any means call transactional memory complicated. In fact it is the easiest approach to concurrency available for those who are used to traditional imperative programming. This is by no means a good thing, since without needing to understand concurrency people will inevitably write "concurrent" programs that won't scale or worse (it's not too hard to imagine a transactional system where performance suddenly drops off a cliff when you add processors because transactions are now being aborted all the time).

A final comment I want to make is to go back to considering objection #1 to transactional memory. While the state-of-the-art in distributed systems research is advancing well, the state of experience in actually implementing them is lagging much further behind, to the point where many people have a very misguided view about what's hard and what's not in programming distributed systems. This applies even to those doing the former, since most distributed systems researchers simply don't have the time to do anything but research (it's a very exciting time for the field, and large rewards will go to those who publish first). This was illustrated in an incident a while ago where a research professor in the field whose work and talents I otherwise highly respect told a class that message passing was too difficult for programmers and that they would prefer to use locks and semaphores to program, so how can we implement those in terms of the former? Of course I had to object. I suspect there are and will be a lot of similar arguments repeated over and over, and the familiarity card will be brought out again and again.

If someone does it to you, point out the fact that out of all software, only a very small amount is written with lock-based concurrency, and out of all those programs, only an insignificant percentage is actually written to be real parallel software (as opposed to multi-threaded GUIs and network servers). In this case, the familiarity card is really worth far less than it seems.

Software Transactional Memory

I promised in an earlier post to discuss software transactional memory, so today I've decided to present a paper written by Andrew Seniuk and me for Lisa Higham's distributed algorithms course. The contains an overview description of STM, discusses implementation approaches of a few current systems, provides proofs of several properties of the DSTM implementation, and highlights possible pitfalls that can occur in transactional code. During the time Andrew and I were working on the paper, there was a lot of excitement in the blogosphere of the potential of STM, and I think it is important to highlight the problems that will be encountered when use of software transactional memory gets more widespread - both to help people avoid them, and to deflate some of the "silver bullet" hype that I think STM has received. Enough rambling, here's the link to the paper:

http://vsedach.googlepages.com/stm561.pdf

Sunday afternoon reading.

Dan Creswell recently put up a "reading list" of recommended articles and papers on distributed system, which you can find here. Some interesting stuff there. The articles about the architecture of MySpace is what finally convinced me to get rid of my MySpace account.

How to write parallel programs

I recently finished reading through Nicholas Carriero and David Gelernter's How to write parallel programs (available free online), a hidden gem on parallel computing published in 1990. Most books about distributed and/or parallel computing that I've come across are banal junk (no, no one needs another book about MPI) - How to write parallel programs contrasts sharply.

I suspect a major reason why parallel programming is considered difficult today is that most people don't have any conceptual feel for the space of possible parallel problem-solving approaches. Most educational material presents a fragmented view of models and paradigms; the vast majority of books focus on a single system or approach. The majority of programmers, not having delved into the breadth of the parallel computing literature, simply are not aware that there exist alternatives that may make their parallel problem-solving tasks much easier.

In light of this, the key insight of How to write parallel programs is distilling parallel programming into three distinct, conceptual paradigms for coordination in parallel problem-solving. It may come as a surprise (I promise I won't provide any more spoilers) that DCOM, CORBA, RMI or any other object-oriented "remote method call" junk is not one of them. Which brings me to a tangent where I rant about the whole idea of "remote method invocation" being beyond stupid by managing to get both message-passing in object-oriented programming and the basic assumptions about distributed systems completely wrong. More people should listen to Deutsch and Kay.

How to write parallel programs is intended to be a textbook, and that means algorithm analysis, which frequently leads to surprisingly practical insights about distributed systems, such as load-balancing considerations, dealing with IO bottlenecks, and even using time complexity analysis to (inadvertently) find faulty nodes. Examples are written in C-Linda (an implementation of Gelernter's tuplespaces). Following the arguments and doing the exercises is well worth the time.

I'm going to end this post with a request for help from the readers: I am trying to track down a copy of a paper referenced in the book: R. Abarbanel and A. Janin. Distributed object management with Linda. Research report, Apple Computer, Cupertino, CA, September 1989. If someone has a copy or can obtain one (any old-school Apple employees reading this?), please get in touch.

Nested Transactions

Here is another post in my series about distributed systems.

This one is about J. Eliot B. Moss's 1985 PhD dissertation on nested transactions, Nested Transactions: An Approach to Reliable Distributed Computing, published by the MIT Press. In it, Moss explains the design of his distributed transaction system supporting nested transactions.

Although Moss's is not the first distributed system to support nested transactions, the dissertations introduces the novel mechanism of two-phase locking and deadlock detection as the mechanism for guaranteeing serializability as a correctness condition.

A contrasting approach to implementing nested transactions is to use multi-version timestamping (introduced in David P. Reed's 1978 PhD dissertation, mentioned in a previous post). This approach is free of deadlock as a result of object timestamping, however it is vulnereable to livelock (slow, long-running transactions retrying repeatedly due to changes committed by quickly running ones). For the same reason, multi-version timestamping does not work well in the presence of a high degree of contention over an object by many transactions. Aspnes et alia's 1988 paper on applying multi-version timestamping to nested transactions provides an extension of Reed's original algorithm that somewhat alleviates this problem.

Nested transactions, while not strictly necessary for many systems, become critical when nondeterminism, such as the 'orElse' construct introduced in Harris et alia's 2005 paper on composable memory transactions and implemented in the GHC software transactional memory system (to be discussed in an upcoming post), is needed.

Croquet system overview

Since distributed and parallel systems and concurrency are on my mind a lot, I thought I would start sharing what I consider to be the "things worth reading" about those topics through this blog (as much as for your benefit, my numerous readers, as it is to help me form my thoughts by committing them (pun fully intended) to blog posts).

To start off is some light reading about the design of Croquet: http://www.croquetconsortium.org/index.php/System_Overview

I think the key things to note about Croquet is how Reed's multi-version timestamping (I plan to make David P. Reed's PhD dissertation a future topic) is applied to messages and extended with the notion of pseudo-real time, how replication of computation is not only seen as not bad but essential to the system, and how actions are performed through a sort of delayed message passing (where the programmer uses the above-mentioned pseudo-real time scheme to specify the delay), which is called "temporal tail recursion." Notice that none of the above features are really orthogonal - none of them would work very well (or at all) by themselves. This is probably the most coherent and well-thought out model for handling time in an interactive distributed system that's out there right now.

Another lesson from Croquet is that security in an open distributed system with mobile code can be handled elegantly at the language instead of the virtual machine level. A powerful example is the fact that temporal tail recursion guarantees that malicious objects cannot go into infinite loops that suck up all computational resources. In Smalltalk all iteration is expressed explicitly as message passing, so there is nothing lost by restricting iteration to temporal tail recursion. This is a demonstration of both the benefits of powerful, uniform paradigms in programming languages, and that of the actor/functional programming paradigm specifically.