Why would you do that, Larry? - Reconciliation, Compensation and Eventual Consistency

In my previous post, I introduced the idea of "Eventual Consistency" and we saw how, due to the CAP theorem, it helps ensure global availability of data by sacrificing its consistency.

In most modern day line of business software development, we are used to dealing with the opposite; "Strong" or "Write" Consistency. If I update "Value 1" in the database to say "Good Morning", then the next time someone reads that value, it says "Good Morning". The system’s state is consistent at the end of a write.

Relational databases such as Oracle or SQL Server are in general "Write consistent" stores. However, if using an “Eventually consistent” such as Azure Table Storage or Amazon SimpleDB, the data is not guaranteed to be consistent at the end of a write, due to the replication delay discussed in the previous post.

These systems will take an amount of time to become consistent. This is referred to as the Inconsistency Window. It’s up to the application to decide how to deal with this inconsistency.

The Reconciliation process can be performed manually or automatically, depending on the circumstances.

The simplest way to demonstrate this as a manual process is by querying a search engine such as Google or Yahoo or Bing. Their indices are "eventually consistent" as you sometimes... brace yourself... come across broken hyperlinks in results! Despite what an Oracle sales person might have you believe, the world does not come to an end if your data is inconsistent. In this case, the end user can reconcile the data manually, by clicking the back button and moving on to the next result. No harm done, maybe a little trade-off in terms of usability, but worth it for access to a vast amount of data.

Removing the requirement for consistent data (by removing locking) means that you perform writes in parallel. Probably the most obvious example of this in action is the auction site eBay. In an online auction, the critical moments are as the end of the auction approaches. You could have a relatively quiet period where people are just watching a series of regular bids, followed by a fairly sudden building of activity as the end nears with many overlapping transactions. A bit like Gustav Holst's "Saturn" which I saw performed live by the RSNO last week. Brilliant! Classical music for nerds!

Anyway, unlike a piece of music, in orchestrating an auction, the order in which bids are written is not actually important. You only need to find the highest bid at reconciliation time. As long as you wait until the inconsistency window has closed, you will have the right answer.

This "right answer" is referred to as the Real state of the system.

When dealing with an eventually consistent store, you need to divide all your data into two kinds of state: Provisional and Real.

Any application transactions should only ever read or write to the provisional state, which is a local view that each of these transactions has - no global viewing allowed! Since there is no shared state, and no global locks, transactions can zoom away unimpeded by each other.

The Real state of the system represents the real state of the world. This always lags behind the provisional state, as the real and provisional states are reconciled periodically normally (although not always) by a background process - in Windows Azure terminology, a "Worker Role".

A solid reconciliation process requires that any actions the sytem performs are revocable. Unlike a transactional system, where we can just call "ROLLBACK", in an asynchronous system some things, particularly those that trigger real-world actions, are "unrollbackable". So how do we revoke them?

We execute what's known in SOA circles as a compensation. An important thing to remember about compensations is that they can be very very expensive. As an extreme example, if I've booked a hotel and discount flight and there are no hotel rooms available, I would have to apply a compensation to cancel the flight, but they might offer no refunds due to the discount. So the compensation cost is not only the cost of the compensation itself, but the sum of both that and any "chained" compensations that it triggers; so, the aim here is to minimise the probability of having to run a high-cost compensation.

The point of reconciliation is that the real and provisional states must be in synch with each other at the end of the process - NOT that they should necessarily contain the same data, but answers to queries against each should return the same thing.

In order to minimize the cost of running any compensations, we may need to do the reconciliation more frequently (at a higher cost).

So once again it's all a balancing act, but the key take aways here are: an understanding of Real vs. Provisional State; and the roles of Reconciliation and Compensation in resolving these states.

NB: Theoretically, distributed systems cannot specify an inconsistency window, as there could always be a catastrophic network failure between geographically distributed replicas. So there are still issues hidden away in there, but in practice most resolve fairly quickly. There are still more patterns for dealing with these things, all based on the not-yet-famous "Well I didn't hear anything" design pattern, also not-yet-known as the “tree falling in the woods” pattern.

Until next time... Soupy Twist

A river runs through it

Hopefully now, you will be becoming acclimatised to some of the best practices for designing large scale systems. Now that we've examined the "application layer" in a bit of detail (implement only Atomic, Idempotent, Stateless operations), we'll move on to look at the "back-end" - the data storage / state / persistence layer.

Fundamentally, all scalable architectures are about replication, be it caching, db replication, or clustering. They are about mitigating the problems of having one central authoritative system which simply does not scale. All because as the number of users and transactions grow, so does contention on a central resource. The only way to deal with this is to add replicas: replica applications, databases, machines, graphics, etc., whatever your application is built of.

Way back in the "DotCom Boom (tm)" days of the internet, Eric Brewer, chief scientist of a company called Inktomi, which ran Yahoo!’s search engine before a young upstart company called Google got their big break by displacing them, came up with an idea known as "Brewer’s Conjecture", or sometimes the "CAP theorem". It basically says that in a system that uses distributed data, you can choose two of the following (never all three):

Consistency, Availability, or Partition-tolerance

Having all three is impossible, as there is a delay that occurs during the creation of a replica. To deal with this delay, you can either lock access to each copy until all are in sync, or put up with the fact that these replicas may be out of synch with each other during the replication process. One of the most interesting demonstrations of this was what used to be known as the “Google Dance”: when Google pushed out their monthly index updates, search results across the globe would all go out of synch (nowadays they use much more frequent updates, and a lot more in the way of personalisation technologies, so you can’t really see this effect any more).

At the compute / application layer, problems of state management are largely solved by adopting the styles of programming discussed previously in this blog (Implement only Atomic, Idempotent, Stateless operations).

However, the most common backend state management system for most web applications is the Relation Database Management System, or RDBMS; usually MySQL, Oracle, or MS SQL Server. These databases provide partition tolerance (to a degree) and a consistent view of data, so logically they must be sacrificing availability.

In his most famous blog post, Werner Vogels, Chief Technology Officer of Amazon, takes the view (and I agree) that not only should we sacrifice consistency for the sake of Availability and Partition-tolerance, but that this is the only reasonable option as a system scales (as the uptime of a whole system is the product of the uptime of all its component parts).

Other people have different views. StackOverflow, a popular programming community site, have chosen to sacrifice availability. You often (in my case a lot! maybe I'm breaking it??) see the message "StackOverflow is currently offline for maintenance. Routine maintenance usually takes less than an hour. If this turns in to an extended outage, we will post details on the blog." This may cause them some interesting challenges as they grow, but certainly at the moment , all things considered, they will not necessarily lose any cold hard cash if their site is down for 20 minutes, whereas Amazon’s losses would be massive.

Buiried deeply in the small print of Amazon’s SimpleDB service, the documentation for the Amazon SimpleDB PutAttributes method has the following note:

"Because Amazon SimpleDB makes multiple copies of your data and uses an eventual consistency update model, an immediate GetAttributes or Query request (read) immediately after a DeleteAttributes or PutAttributes request (write) might not return the updated data."

Windows Azure’s Table Storage and both Amazon and Azures Queue Services also share this characteristic, and so cannot guarantee message ordering (FIFO).

This model of "Eventual Consistency" is also one of the core characteristics of the new breed of NoSQL databases appearing. And whilst this makes a lot of sense to many Web Developers who have spent their lives dealing with stateless cached data, it's probably the single most confusing part of distributed application development to a traditional RDBMS or Client/Server developer.

The very idea! Eugh! Yuk! You mean you can update data only for the system to "sometimes give you the wrong result". This makes RDBMS-centric developers feel very nauseous. But fear not! Eventual consistency is part of daily life. As I've said in a previous post, it's odd that the canonical example of the transaction is the bank transaction, when in fact this process operates in an eventually consistent manner (and is actually far more complicated than a transaction).

We've been indoctrinated by the Big RDBMS vendors, that only transactional write consistency will do. They commonly use the banking metaphor to imbue their products with an implied stable, trustworthy state that is simply not there.

People just do not need to know exactly when someone else does something, just what they've done, when they come to deal with it. The mother of all transactional systems, the stock-exchange, is eventually consistent. However a LOT of money is spent to get the inconsistency window as short as possible, for example, locating trading offices as close as possible to exchanges with high speed connections.

There is absolutely no doubt that dealing with an eventually consistent store is much more complicated than dealing with a central RDBMS, but as multi-tenant systems become the norm over the next 5-10 years, the RDBMS will look increasingly creaky as a core on which to build an application.

Over the next couple of posts I'll look in more detail at eventually consistent data stores and queues and some patterns commonly used to deal with it.

Intermission

Please stay tuned for some important messages.

I've been on holiday this week. So you'll just have to wait a bit for my next thrilling post :-).

In the meantime... Some interesting holiday reading...

http://research.microsoft.com/en-us/um/people/horvitz/leskovec_horvitz_www2008.pdf

And this made me laugh. ..

Normal service will resume shortly.

Unleash the beasts!

Following my last post on caching patterns, I was asked a question about how to scale cache repopulation. The question was "In a read through caching scenario... assuming your cached object takes 30 seconds to calculate, and you have 60 processes accessing the cached value every half a second each. This suddenly expires, resulting in 120 misses per second until the value is re-calculated and re-cached successfully, swamping your server with repopulation requests. Right?" Well, erm, yes. Yes it does. Within some of the more nerdy circles, this is often called the Dog Pile effect, and has brought many websites down.

When caching quick-to-create items, this rarely causes problems. However in the case of large data sets that take a while to calculate, this can prove "the straw that breaks the camel’s back".

From Wiktionary: v. to dogpile

1. To jump on top of someone, usually in a group.
   2003, Nancy Holder, Buffy the Vampire Slayer: Chosen[3], ISBN 0743487923, page 657:
   A vampire got her around the neck from behind; then more, dogpiling her.
2. To pile on or overwhelm, such as with criticism or praise.
   2005, Craig Spector, Underground[4], ISBN 0765306603, page 169:
   But this guy was serious, using online payment services and dogpiling her e-mail box within minutes, requesting expedited shipping.

You shouldn't really have anything in your system taking so long to repopulate. If you do, it's likely to indicate a more fundamental architectural problem, as whatever you’re caching does not have a high degree of disposability.

This is a situation where we have hundreds of workers looking for the same information, so we don’t want to swamp our data source with the requests for that same information.

What we want to do, in theory at least, is use a lock. OMG. I know. I said never use locks, but here we are using them. What gives?

Well, this is really an optimisation for a particularly unusual case. Here, cached items are immutable, so we have no side effects to worry about. The process for repopulating lists is just not very efficient. Locks CAN be used in a distributed system, but just very carefully (and normally on a singly responsible background process). For example, some data reconciliation processes may require locks on data, but by putting these processes on background workers, we try to ensure they do not block user activities.

You also need to be aware of the problems of using locks in your particular environment. (We're OK here as our write operations are idempotent).

In this case we would write a wrapper around the cache, to check on its expiry, and force only a single thread to repopulate the cache.

So in pseudo code...

function GET(key)
{
  object = Cache.Get(key) //try and get the value from the cache
  if (object is null)
  {
    lock //SERIALISE ACCESS
    {
      if (object is still null)
      {
        object = GetItemToCache() //takes 30 seconds
        PUT(key, object, now + 15 minutes) //expires in 15 minutes time.
      }
    }
  }
  return object
}

function PUT(key, value, expires)
{
  Cache.Insert(key, value, expires)
}

Now you still have processes queuing up at the point where access is serialised, so in the 30 seconds required to repopulate the cache, you have 120 processes per second blocking. This solves one problem but creates another.
We've offloaded the traffic from GetItemToCache() , but forced everyone to wait. So better, but still not too great. How can we improve this to avoid blocking even further? Well... we can temporarily re-cache the "expired" value!

Here, we know that our cached object takes (about) 30 seconds to repopulate, so we can estimate when the item will go "stale". When the data goes stale it expires. This allows users to continue to read stale data whilst the update is prepared. avoiding contention on the object loading process.

.NET's cache for example allows you to specify a callback, which runs when an item falls out the cache. Most caching platforms have a similar mechanism for handling dropped cache items.

Using ASP.NET expiry, callback functions are delegates that support the following signature:

void CacheItemRemoved(string key, object val, CacheItemRemovedReason reason);

So our newly refactored pseudocode is...

constant ItemKey = "Item_That_Goes_Stale"; //a key to identify the cached item
function GET()
  {
    object = Cache.Get(ItemKey) //try to get the item from the cache.
    if (object is null)
    {
      lock //if we have a miss, we synchronise access to repopulate the cache here.
        object = Cache.Get(key)
        if (object is still null) //double check after a lock (another thread could have jumped in.)
        {
          if (object is null) POPULATE_CACHE(); //If it's not in the cache.. cache it.
          return object
        }
      end lock
    }
  return object
}

function PUT(value, expires)
{
  //Here we add a callback to retain the value in the cache in the case of expiry.
  //ASP.NET will call MAKE_STALE just before the item is dropped from the cache.
  Cache.Insert(ItemKey, value, expires - 30 seconds, MARK_STALE);
}

function MARK_AS_STALE(string key, object val, CacheItemRemovedReason reason)
{
  if (reason for removal is Removed or Underused) then return;
  // do no re-cache the item if it's removed (written over) or underused.
  //cache is stale... repopulate it straight away... Now if someone calls GET in the next 30 seconds they will get the old value.
  Cache.Insert(ItemKey, value); //this value will be removed when the following POPULATE_CACHE call replaces it.
  POPULATE_CACHE(); //REPOPULATE THE CACHE
}

function POPULATE_CACHE()
{
  object = GetItemToCache() //takes 30 seconds
  PUT(key, object, now + 15 minutes) //expires in 15 minutes time.
}

If you are not using a platform which supports this type of callback pattern, you can use the cache itself to check for staleness by caching a "still fresh" flag for an object which expires shortly before the real item. You can then check the cache for both the real item and the "still fresh" item. If the "still fresh" flag is not in the cache, we know the real item is stale. This does require a double read of the cache.

Either way, your application should remain much more responsive and stable if this technique is employed to freshen the cache ahead of time. Again, I must say this is an optimisation that is hopefully not needed too often. If you’ve designed your systems around distributed architectural principles, you should never need to do this. The vast majority of the time, the standard read/write, through/behind caching techniques should be more than enough to achieve acceptable system responsiveness.

The central principle and widespread use of caching in large scale web applications, and the new post-relational NoSql databases (used in the likes of Amazons Web service platform and Microsoft Azure Queues), lead to some interesting and common problems with data consistency which I will start to look at in my next post.

Warning: This internet may contain traces of nuts

To summarise my last few posts, you should now have at least a basic understanding of how the web is put together (DNS, TCP/IP, HTTP, HTML) and the problems you are likely to encounter when writing code for distributed systems. You should also have an idea of the major architectural constraints to consider when creating robust code in this type of environment....

• Atomicity - Write atomic operations that pass or fail in their entirety.
• Statelessness - Maintain a high degree of disposability
• Concurrency - Avoid using locks wherever possible.
• Idempotence - Write code that is repeatable and has no side effects

Great. We can now write a working distributed application. However, while this application may be effective, it may not be the most responsive. All this data still has to move all over some very s_l_o_w wires. In all distributed applications, there is a rule of thumb that you should remember. Data is always more expensive to move than it is to store. You only have to look at mobile phone operators data tariffs to see that.

So, how do we stop moving stuff around unnecessarily? Well, the answer is actually modelled directly on the world of squirrels. Well maybe not directly. Both squirrels and distributed applications use caches for storing stuff around their domain until it's needed.

In practical terms most caches have interfaces like a hash table, so you effectively have a big bucket in which you can store objects and identify them using a key. You can add an item to a cache by calling PUT(key, objectToCache) and retrieve items by calling GET(key)

Most caches are of limited size, so we have to decide how to throw things away if the cache is full and we want to add new items. This is normally done by popularity. So those items "hit" the most stay in the cache, and less popular items are thrown away.

Also in common with squirrels, we have an issue with staleness. We don't want to be retrieving nuts only to find that they've gone all mouldy. Yuk! So most caches also support some sort of Data Expiration strategy. These can be defined in a number of ways, though normally they tend to support sliding and absolute expiry policies. The former is where you specify after what period of inactivity you want cached items to expire. i.e. "If this isn't used for 5 minutes throw it away". The latter, absolute expiry is just the same as a use-by date..."this data is no good after midnight on wednesday."

Now an important thing about caches, is that you should not make any assumptions about their consistency. They could have been pillaged by some other hungry squirrel in the middle of the night without your knowledge.

As the cache makes decisions about how to optimise its use, it is quite possible that you can call PUT(itemA, myItem) and then call GET(itemA) and you get NULL back.

In technical jargon...
If you call GET(itemA) and it returns myItem, this is called a "hit"
If you call GET(itemA) and it returns NULL this is called a "miss".

Caches come in all shapes and sizes:
In-Memory Caches, Distributed Caches, Proxy caches, and Large Area Caching Networks , but they all work in more or less the same way.

So how do we make best use of a cache? Essentially, it's all to do with understanding the cost of a miss. You take the slowest operations.

There are effectively 4 caching patterns that you see on a day to day basis: Read Through, Write Through, Read Ahead and Write Behind.

For reading, it's all about how you handle a cache miss. Using the Read Through pattern in the case of a cache miss, you query your data source, populate the cache, and continue. Next time you get a hit.

With Read Ahead you should never get a miss. You query your data source and populate the cache before any requests are expected. But you may be holding things in memory that are not required.

Populating a cache on a write is more unusual outside distributed applications, but can again boost performance given the right set of conditions. Using Write Through caching, whenever a user submits data, this data is cached as it is written to the data source, speeding up any following reads without having to query the data source again.

And finally, Write Behind. Now, this one is seen particularly often in distributed systems. Most of the "post-relational" databases out there use this pattern, in which the cache is firstly committed to, then some other process writes to the data source at some unspecified later time. This has the advantage of handling writes very very quickly, and leaves the complicated business of reconciling data until later. However this means you have to tolerate a degree of inconsistency across nodes, albeit momentary in many cases.

So what pattern should I be using? Well, as usual there is no straightforward answer. "It all depends".

Read-Ahead gives a speed advantage over Read-Through, but can only be used if you can predict which cache items are going to be needed in the future. If your guess is accurate, refresh-ahead means reduced latency and no added overhead. The poorer your guess is, the worse your performance will be (as more unnecessary requests are sent to your data source). This could even result in worse overall performance if your data source falls behind with its request processing backlog. Read Ahead caches are also more expensive to initialise.

If Write-Behind caching can be used, it can allow much lower latency and higher throughput than Write-Through caching. Additionally Write-Behind caching can sometimes improve the performance of the data source (as there are less writes, and they can be delayed until resources are available).

So, caching can be used to improve the efficiency of distributed systems. You often see caching used like some kind of silver bullet for speed, but like any form of optimisation, it should be viewed as such. The trade off for speed is use of more resources (and management of those resources). More components mean more complexity, but applied carefully, caching can greatly improve the responsiveness of an application, especially a distributed one.

Do it Again!

This is the last in my mini-series about the fallacies of network computing before moving on to deal with the practical side of dealing with these things.

The final fallacy to address, states that "the network is reliable". As you know if you ever used AOL in its, erm... heyday? - it's not. It may go a bit faster now, but it's the same old network that it used to be.

What kinds of operations can be safely performed on something so unreliable? The answer is: "Idempotent" ones. Gosh, Latin! How old is this internet?

idem - meaning "the same". You'll see this used in legal papers abbreviated to "id." as lawers do tend to repeat themselves a lot, and are by far the biggest fans of dead languages (outwith Delphi programming circles).

potent - meaning power

So: "having the same power". But what does that mean? Essentially it's a big fancy word for something that's very simple (in theory at least).

Idempotent operations are essentially repeatable, and have no side effects.

Due to the infrastructure of the net, it is often the case on the web that messages between end points may be sent and received more than once, so it's important that messages processed multiple times do not cause side effects. The clichéd example of this is the browser refresh, and while this is an example of idempotence, I feel it's much clearer to explain by showing what it's not.

Voting systems are the common example of failing to take idempotence in to account.

Even though the "replay attack" is one of the oldest forms of internet abuse, it is still surprisingly common to come across sites that contain polls which are easily rigged. Essentially the web page presents "Vote for A" and "Vote for B" as functions. You click "Vote for B" and a counter is incremented. If you fire the request again, the counter is incremented again.

Write...

for (int i = 0; i < 10000; i++) { VoteForB(); }

... and zap! In a fraction of a second, "B" becomes the clear winner.

Increment() IS NOT an idempotent function. It has side effects on application state and is not repeatable.

An idempotent operation is both repeatable and passive.

So, a toggle() function is not idempotent. A Sqaure() Function or the evaluation of a regular expression against given input is idempotent. It does not change any value given to it and has no side effects.

A * A will always equal A squared.

You can of course deconstruct many non-idempotent operations in to a number of idempotent equivalents, e.g. SwitchOff() and SwitchOn() functions rather than Toggle().

Calling SwitchOff() SwitchOff() SwitchOff() SwitchOff() SwitchOff() SwitchOff() will always result in the Switch being Off.

Idempotence is often confused with method safety. A "Safe" method has absolutely no side effects, for example a read operation is Safe. A Delete() operation is not "Safe" but it can be made to be idempotent. i.e. repeatable with no side effects.

This type fault tolerance is actually designed in to HTTP itself. See:
http://www.w3.org/Protocols/rfc2616/rfc2616-sec9.html#sec9.1.2

Fundamentally, the RESTful-style of the web is different from the CRUD style of most database-oriented systems.

Under RESTful systems using HTTP, calling PUT multiple times will result in the resource existing at a particular address. Calling DELETE multiple times will result in the item no-longer-existing.

You can call PUT PUT PUT or DELETE DELETE DELETE and the end result will be the same.

The comparable operations using something like SQL will result in errors beyond their first call: "The record already exists" or "The record does not exist", respectively.

On the web, you ignore idempotence at your peril. Famously, Google Web Accelerator wreaked havoc and was quickly withdrawn due to web developers widely ignoring idempotence and method safety.

Almost overnight, millions of web developers learned about idempotence the hard way.

While idempotence seems fairly simple on the face of it, the reality of it can quickly become very, very complicated, particularly since, in most distributed systems, you need to take in to account the idea of sequential idempotence. If you're looking at a system with multiple workers of some sort, either threads, VM instances, or WorkerRoles in Azure, and if you're using lock-free structures (as you should be), the possibility exists that you may have to process the same message not just more than once, but also out of order.

For example, suppose that someone updates a record, while another simultaneously deletes it.
Here, I have an Idempotent PUT and an idempotent DELETE verb. I can call PUT PUT PUT or DELETE DELETE DELETE, and have the same end result. But now I have an extra problem: I now require two independent operations to have combined idempotence.

The following sequences (and even all of these if run in parallel together)

PUT DELETE
PUT PUT DELETE PUT DELETE PUT
DELETE PUT

should all have the same end result. What is that end result?

Now you can see how idempotence works, you can think about how to solve this. And if you want a clue see this article:
http://www.spamdailynews.com/publish/Google_subpoenaed_to_turn_over_deleted_Gmail_messages.shtml
The important thing to remember is to write code that is repeatable and has no side effects. The result of this is that you enable fault tolerance, using the simple policy of "If you're not sure it worked, just do it again".

For Sale: One slightly worn RDBMS

Web and cloud apps have to be able to scale as close to linearly as possible, in a hostile environment. Network topology and number of users can fluctuate wildly. Contrast this with a desktop application, which just has to handle a single user. Or with a client/server app, which may have to handle a couple of hundred users usually on the same network. If millions of users start using our distributed system, the experience of each should be no different from what it would be if only 5 or 6 people were using it.

While web developers are used to dealing with concurrency, either consciously or just by knowing that calling Application.Lock() from VBScript will cripple their app, it is also becoming a wider issue for desktop developers, as chip manufacturers squeeze more and more cores on to their chips.

The arch-enemy of concurrency is shared data. Any data that two people have to update can cause concurrency problems. This is actually made slightly simpler in web programming by virtue of HTTP being stateless, and is one reason why the functional style of programming as seen on the web is becoming more popular in other applications.

If we can maintain the qualities of statelessness and atomicity in our operations, we are already well on our way to make our code safe to run concurrently. It's only when someone attempts to force a stateful model that you run into problems.

If shared data is the enemy of concurrency, locks are its atomic bombs. Locks work by serialising access to shared data, so that only one process / thread / fibre can write shared data at any one time. This means that anyone else just has to wait in line. Serialisation is the arch enemy of scalability.

So, NEVER hold editable shared data in your application layers. I will of course break this absolute rule in a future post for grandstanding purposes, but I'm allowed to - It's my blog! And of course, it's to illustrate a good point.

This issue doesn't tend to be a problem for those accustomed to web programming, since most web frameworks either enforce, or heavily imply, avoidance of global state. In most real systems today, shared state is handled by a backend RDBMS, such as Oracle or SQL Server, which offer a gigantic improvement over file-based DBs such as Access or Paradox. These do a very good job, at the moment, of handling medium to large scale apps. However, if we scale to the web, where fast global availability is required, then "Houston, we have a problem." We realise all we’ve done is offloaded the problem further down the chain. Because these client-server style databases have transactions, and therefore locks, embedded firmly at the heart of their architecture, they will not scale globally. You can partition them; you put them on ever larger machines, or even clusters of machines; but they will always require a global lock to perform a commit.

So if RDBMS' such as SQL Server don't scale well, then what does?

Whilst still a drop in the ocean in terms of commercially enterprise databases, Databases based on Distributed hash tables, such as CouchDB, Google’s BigTable, Voldemort, AWS SimpleDB and similar, are gaining a lot of traction.

Microsoft’s forthcoming Azure Table Storage falls in to this category too, unlike their slightly clumsy and poorly positioned "old man in a new hat" - SQL Azure Services.

These types of database characteristically emphasise Decentralization, Scalability and Fault Tolerance, with nodes continuously joining, leaving, and failing within the system.

Sound familiar? Well these are all databases designed to work not only on the web, but in accordance with the fundamental architecture of the web. They emphasise availability over transactional consistency which can cause serious headaches for people stuck in an RDBMS world view. (The idea that the database will sometimes return the wrong result is horrifying to them.)

Scalable Transactions are EXTREMELY complex (actually impossible, as previously discussed) in a distributed system. So, support for transactions in Distributed Databases is rare to limited. If you look at Amazon's SimpleDB or Azure Table Store, if you perform a "BatchPut" or "EntityGroupUpdate" respectively, these do succeed or fail as a unit. However, they are subject to limitations well below the expectations of developers who are used to developing against RDBMS systems. Maximum request sizes of 4MB, or number of items in a transaction (25 for SDB), seem almost laughable. However these distributed document style databases are a fundamentally different type of application from the RDMBS.

Unlike SQL Server or Oracle however, they are reliable, cheap, and MASSIVELY scalable. Therefore, while they account for a tiny fraction of the database market at the moment, they do offer a very serious cost-based competitive advantage over large scale installations of RDBMS systems. I will write a future post about lock free data management patterns and post-relational DBs in the future, but once again, for now the key takeaway for concurrency is:

Locks are mindblowingly expensive in a distributed system.

And they get more expensive the more they are used. If you really MUST use them, stand well back and wear very dark sunglasses.

Next and last up: Idempotence