Showing posts with label Eventual Consistency. Show all posts
Showing posts with label Eventual Consistency. Show all posts
Tuesday, 10 November 2009
To be doomed to know the future and not to be believed
Just came across this. Digg are using Facebook's (now Apache's) "Cassandra" Distributed Datastore. Interesting.
Thursday, 5 November 2009
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
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
Thursday, 1 October 2009
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.
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.
Subscribe to:
Posts (Atom)