Will smart memory need an ACID model?

This is really part II of my posting on smart memory from ten days ago.

Let's imagine that we've created the ultimate data warehouse, using Derecho.  This warehouse hosts terabytes of persistent memory, It can absorb updates at a staggering rate: hundreds of gigabits per second, has tens of thousands of processors that crunch the data down and then store it as collections of higher level "knowledge models", and is always hungry for more.  

We used to think of data warehouses as respositories for nuggets of data, so perhaps we could call these "muggets:" models, that can now be queried.  Hopefully this term isn't painfully cute.

Next, we imagine some collection of machine learning applications that consume these muggets and learn from them, or compute some sort of response to them.  For example if the muggets represent knowledge collected on the local highway, the queries might be posed by smart cars trying to optimize their driving plans.  "Is it ok with everyone if I shift to the passing lane for a moment?"  "Does anyone know if there are obstacles on the other side of this big truck?"  "What's the driving history for this motorcycle approaching me: is this guy some sort of a daredevil who might swoop in front of me with inches to spare?"  "Is that refrigerator coming loose, the one strapped to the pickup truck way up ahead?"

If you drive enough, you realize that answers to such questions would be very important to a smart car!  Honestly, I could use help with such things now and then too, and my car is pretty dumb.

We clearly want to answer the queries with strong consistency (for a smart car, a quick but incorrect answer might not be helpful!), but also very rapidly, even if this entails using slightly stale data.  In Dercho, we have a new way to do this that adapts the FFFS snapshot approach described in our SOCC paper to run in what we call version vectors, which is how Derecho stores volatile and persistent data. Details will be forthcoming shortly, I promise.

Here's my question: Derecho's kind of data warehouse currently can't support the full ACID database style of computation, because at present, Derecho only has read-only consistent queries against its temporally precise, causally consistent snapshots.  So we have strong consistency for updates, which are totally ordered atomic actions against sets of version vectors, and strong consistency for read-only queries, but not read/write queries, where an application might read the current state of the data warehouse, compute on it, and then update it.  Is this a bad thing?

I've been turning the question over as I bike around on the bumpy, potholed roads here in Sebastopol, where we are finishing up my sabbatical (visiting my daughter, but I've also dropped in down in the valley and at Berkeley now and then).  Honestly, cyclists could use a little smart warehouse help too, around here!  I'm getting really paranoid about fast downhills with oak trees: the shadows invariably conceal massive threats!  But I digress...

The argument that Derecho's time-lagged model suffices is roughly as follows: ACID databases are  hard to scale, as Jim Gray observed in his lovely paper on "Dangers of Database Scalability".  Basically, the standard model slows down as n^5 (n is the number of nodes running the system).  This observation gave us CAP and BASE and ultimately, today's wonderfully scalable key-value stores and noSQL databases.  But those have very weak consistency guarantees.

Our Derecho warehouse, sketched above (fleshed out in the TOCS paper we are just about to submit) gets a little further. Derecho can work quite well for that smart highway or similar purposes, especially if we keep the latency low enough.  Sure, queries will only be able to access the state as of perhaps 100ms in the past, because the incoming database pipeline is busy computing on the more current state.  But this isn't so terrible.

So the question we are left with is this: for machine learning in IoT settings, or similar online systems, are there compelling use cases that actually need the full ACID model?  Or can Maine learning systems always manage with a big "nearly synchronous" warehouse, running with strong consistency but 100ms or so lagged relative to the current state of the world?  Is there some important class of control systems or applications that can probably be ruled out by that limitation?  I really want a proof, if so: show me that omitting the fully ACID behavior will be a huge mistake, and convince me with mathematics, not handwaving... (facts, not alt-facts).

I'm leaning towards the "Derecho would suffice" answer.  But very curious to hear other thoughts...

Will smart memory be the next big thing?

We're fast approaching a new era of online machine learning and online machine-learned behavior: platforms that will capture real-time data streams at high data rates, process the information instantly, and then act on the output of the processing stage.  A good example would be a smart highway that tells cars what to expect up around the next curve, or a real-time television feed integrated with social networking tools (like a world-cup soccer broadcast that lets the viewers control camera angles while their friends join from a remote location).

If you ask how such systems will need to be structured, part of the story is familiar: as cloud-hosted services capturing data from sensors of various kinds (I'm including video cameras here), crunching on it, then initiating actions. 

A great example of such a service is Microsoft's Cosmos data farm.  This isn't a research platform but there have been talks on it at various forums.  The company organized a large number of storage and compute nodes (hundreds of thousands) into a data warehouse that absorbs incoming objects,  then replicates them onto a few SSD storage units (generally three replicas per object, with the replication pattern fairly randomized to smooth loads, but in such a way that the replicas are on fault-independent machines: ones that live in different parts of the warehouse and are therefore unlikely to crash simultaneously).

Once the data is securely stored, Cosmos computes on it: it might compress or resize an image or video, or it could deduplicate, or run an image segmentation program.  This yields intermediary results, which it also replicates, stores, and then might further process: perhaps, given the segmented image, it could run a face recognition program to automatically tag people in a photo.  Eventually, the useful data is served back to the front-end for actions (like the smart highway that tells cars what to do).  Cold but valuable data is stored to a massive backend storage system, like Microsoft's Pelican shingled storage server.  Unneeded intermediary data is deleted to make room for new inputs.

Thus Cosmos has a lot in common with Spark, the famous data processing platform, except on a much larger scale, and with an emphasis on a pipeline of transformations rather than on MapReduce.

If we step way back, we can start to perceive Cosmos as an example of a smart memory system: it remembers things, and also can think about them (process them), and could potentially query them too, although as far as I know Cosmos and Spark have limited interactive database query functionality.  But you could easily imagine a massive storage system of this kind with an SQL front-end, and with some form of internal schema, dynamically managed secondary index structures, etc.

With such a capability, a program could interrogate the memory even as new data is received and stored into it.  With Cornell's Derecho system, the data capture and storage steps can be an asynchronous pipeline that would still guarantee consistency.  Then, because Derecho stores data into version vectors, queries can run asynchronously too, by accessing specific versions or data at specific times.  It seems to me that the temporal style of indexing is particularly powerful.

The interesting mix here is massive parallelism, and massive amounts of storage, with strong consistency... and it is especially interesting that in Derecho, the data is all moved asynchronously using RDMA transfers.  Nobody has to wait for anything, and queries can be done in parallel.

Tomorrow's machine learning systems will surely need this kind of smart memory, so for those of us working in systems, I would say that smart memory architectures jump out as a very exciting next topic to explore.  How should such a system be organized, and what compute model should it support? As you've gathered, I think Cosmos is already pretty much on the right track (and that Derecho can support this, even better than Cosmos does).  How can we query it? (Again, my comment about SQL wasn't haphazard: I would bet that we want this thing to look like a database).  How efficiently can it use the next general of memory hardware: 3-D XPoint, phase-change storage, high-density RAID-style SSD, RDMA and GenZ communication fabrics?

Check my web page in a few years: I'll let you know what we come up with at Cornell...

The ultimate Paxos (and Atomic Multicast) protocol

Many years ago, probably around 1986, Barbara Simons organized a workshop on replication at Asilomar, a California resort that had an iconic role in early systems research (it took a while, but eventually our community was too large to fit there). 

Her idea was to bring people together across areas: theory and practice, and within the practitioners, databases as well as other forms of distributed systems.  It was a good idea, and a great workshop.

In Jim Gray's talk, he pointed out that the interaction pattern we associate with the 2-phase commit protocol was seemingly universal, and that perhaps we were wrong to think of 2-phase commit as primarily a database construct.  Instead, he then asked, if 2-phase commit is really concerned with something deeper than database transactions, what is this deeper core concept?  The essential point of his talk was that the 2-phase pattern seen in 2PC clearly allowed a distributed system to "learn" something.  And from this, he suggested, that if  we could start to appreciate the minimal "knowledge" required for a distributed system to be correct, we could build a minimal implementation of that protocol and solve the problem once and for all.

A long time has passed.  Jim is no longer with us, and we know all about knowledge in distributed systems, and how processes learn.  Even so, Jim's question still sometimes comes up.  I mentioned this to Idit Keidar today over lunch, and her response was definitive: she convinced me that today, we really do (finally) understand the answer to Jim's question.    Her thinking centers on work she published in 2006 with Alex Shraer.

Here's what she explained.

First, she observed, distributed computing is ultimately about just one question: fault-tolerant consensus.  For her, Jim had noticed this, but was understanding it as something about the 2PC communication pattern rather than appreciating that the thing that matters more is consensus: moving from a state in which some decision is uncertain to one in which a decision has been reached.  As Idit views the question, one can transform consensus into other forms, we can route the messages in all sorts of patterns, but the bottom line is that either we achieve agreement via consensus or we simply aren't building systems that can make logically sound claims about their behavior. 

Next, she pointed out, agreement is trivial while failures aren't happening.  While a system remains stable, the author of a new piece of information simply needs to multicast it, and this accomplishes consensus.    If there might be many authors, you can circulate a token around them either before they send (as in the old Totem and Transis protocols) or after the messages arrive, to assign them a delivery ordering.

And so in Idit's view, the only hard issue is to handle failures.  Obviously, in the case of 2PC, particularly back in 1986, the field started out with a somewhat muddled understanding of this aspect: 2PC starts by stating that there is a set of processes that need to achieve agreement, but where does that set come from?  In a modern system like Derecho, the set itself is the output of a consensus layer.  2PC uses a crash failure model, but this is one of many models.  Back in 1986 people were very focused on models: crash failures, failstop failures, omission to send, omission to receive, timing errors, Byzantine models.

But these days all that really matters is to handle crash failures correctly.  Some organizations toy with Byzantine fault-tolerance, but honestly, I've never met anyone who had a service that actually came under a Byzantine attack.  Maybe the blockchain people will finally have that experience.

So let's focus on crash failures for a moment.  In light of the FLP theorem, we know now that without a perfect failure detector, protocols can't be proved correct: you can show safety, but not liveness.  In practice, we solve this by forcing a seemingly faulty process to leave the system and then, if you want, it can rejoin. 

So failure becomes a purely behavioral abstraction: if the majority of the system deems some process to be worthy of excluding, for whatever arbitrary reason it desires, than out goes the offending process, end of story. 

Where did the majority constraint come from?  The role of the majority restriction is to prevent logical partitioning. 

So, we end up with a rather dynamic form of membership: the system defines its own membership, excluding processes that seem faulty, and makes progress as long as it can maintain a dynamic form of majority.  To whit: at any point in time, some epoch defines the composition of the system.  In order to move to a new epoch, the systems needs agreement by the majority of whoever is in the active epoch.

So here's where all of this leads: Derecho is a concrete implementation of this new-age perspective on Jim's question.  Indeed, it is the ultimate story in the sense that the system is optimal in many different ways.

To appreciate this, you'll need to read the Derecho papers, but in a nutshell, the system maps consensus onto RDMA hardware in an exceptionally efficient way.  But the protocol itself runs consensus on the configuration of each epoch, and then uses a cheaper consensus protocol within an epoch (one that can assume membership is stable and that failures won't occur), leaving a very clean realization of distributed computing.

Indeed, Derecho is a constructive lower bound in every interesting dimension.  It is an optimal solution to the problem of distributed computing, using consensus-based methods.

Why do I make this claim?  Well, first, during a given epoch, the system is as quick to deliver messages as is possible: one can prove that any system that delivers messages with fewer "exchanges of information" between its processes is either incorrect, or at last at risk of needing to stop because of some single fault.

Next, one can show that Derecho's agreement on the state of each membership epoch is optimal.   Here, Derecho makes use of an all-to-all pattern of information exchange, and I asked Idit if she thought the protocol could be improved.  Idit pointed again to her 2006 papers with Alex.  Without prior knowledge of which processes failed, and how many failed, she explained, this pattern of information exchange is the quickest way to reach agreement on the membership of the next epoch.

Finally, we can show that Derecho makes progress with a failure model called <>P: eventually perfect failure detection.  Idit explained to me that in the past, people actually thought it might make sense to focus on progress with weaker detection models, like <>W, but that if you do so, you either end up with a fault model equivalent to <>P, or you end up with slower agreement protocols that look much more like Byzantine agreement.  So, for the style of quick progress Derecho is after, she explains, <>P is the right goal.  And Derecho is live with <>P, provided that no more than a minority of processes fail in any epoch.  Which again, is the best one can do.

Now, Idit is a theory person, but it is interesting for me as a practitioner to also think about practicalities.  As the old adage goes: "In theory, theory and practice are the same, but in practice, they differ."

As it happens, Derecho is optimal in a communications-engineering (networking) sense. 

Given that reliable networks always have a one-to-one acknowledgement based layer, reliable multicast over a tree is an optimal data dissemination pattern: if you try and disseminate multicasts over an unreliable 1-N protocol, the cost of detecting lost packets and resending them will be very high compared to a tree of unicast transfers.  (Perhaps optical networks with optically aggregated acknowledgements could offer a slight hardware speedup, but even this isn't clear, since the optical aggregator would itself be a tree). 

Now if you happen to be a cloud computing engineer, and read the above, what will jump to mind is heavy-tailed behaviors and multi-tenancy: Derecho protocols sometimes relay data, so if a relayer is very slow, everyone waits.  Paxos classically would avoid this using quorums: the system doesn't wait for the slowest process.  So how can I claim that Derecho is optimal in a practical sense if it doesn't use quorums?

There are a few answers.  A long-winded one would talk about work Fernando Pedone and his student, Parissa Jallali, did while visiting my group a few years ago.  Basically, they studied quorum behaviors in Paxos and discovered that while Paxos can briefly "outrun" a slow process, eventually work piles up and either flow-control causes the protocol to pause, or a shifting pattern of just who is the slow process switches the slow guy into the quorum and some previous quorum member becomes slow.  Either way, Paxos basically halts until everyone catches up and is back in sync.  So quorum patterns do not evade the disruption caused by heavy tailed behaviors.

Conversely, offloading the protocol into hardware actually can eliminate that issue, because the hardware is dedicated: the network spends 100% of its time communicating, so if you can describe a pattern of communication, then let it rip, the network is the ideal engine for moving data.  As it happens, Derecho generates deterministic data transfer schedules, hence given adequately programmable NICs we can hand the entire sequence of block transfers to the NIC.  So we can even make "optimal" use of the network, and since a NIC never sleeps or pauses, quorum behaviors aren't needed even if end-applications sometimes are a bit slow.

So a road that for me started around when Jim asked his question about 2PC seems to have reached its end: Derecho implements the ultimate Paxos protocols for atomic multicast (often called vertical Paxos) and for persisted memory (classic Paxos).  We could add an optimistic early delivery protocol with a flush, too, as in Isis and Vsync, but we decided to keep the system simple and omitted it: most people who would use that feature probably just want raw RDMC from the Derecho API, and this we do offer.

And so the Paxos problem is solved.  And you know what?  Its about time!  Idit feels that it was solved back in 2006, actually.  As for me, well, until I can write an application using the ultimate protocol, the problem is open.  But today, I finally can.  (I don't mind at all that Derecho employs a protocol that is actually extremely similar to our first Isis protocols from 1985.)

So should we all pack our bags and go home? 

Not quite yet.  First, there are other forms of distributed consistency, like convergent (gossip) protocols, self-stabilization, and of course, Byzantine Agreement.  It would be nice to see how those fit into this picture and whether there can be a single integrated story that combines all the elements.

A second issue is the engineering complexity of modern platforms.  I'll write a whole blog posting on this sometime soon, but suffice it to say that in a data center with physical topology (racks, switches, TOR switches, failure-independence domains...), GPU clusters, NetFPGA accelerators, Intel SGX protection enclaves... it just isn't obvious how to write code for such environments.  Derecho is just part of the answer, and not the whole story.

Then beyond all of this are dimensions we have yet to tackle in Derecho itself.  For example, even if Derecho is the ultimate data center protocol, is it also the ultimate WAN version?  As it happens, I suspect that this may be true too, but more attention to the question will be needed.  Anyhow, until we have it running, I won't believe the story even if I figure it out "in theory".  After all, in theory, theory and practice are the same...  but in practice, they are enormously different.

So I wouldn't despair: very likely we are finally at the end of the road for Paxos, but it certainly isn't the end of the road for distributed systems.  Watch this space: I can promise plenty of interesting new questions, and new answers, in years to come.

Transactions [4]: How does a system like Derecho fit in a world with transactional DHTs?

Derecho is Cornell’s new RDMA-based data replication solution.  It is a simple library coded in C++ 14 (well, maybe not that simple, but simple to use) and it automatically interfaces your code to the local RDMA hardware.  It also automates such tasks as initializing new members when an application process joins an active application or restarts after a serious crash (state transfer), membership coordination and update when members join or fail, persistent data management, atomic multicast, and even RPC (but we also allow you to interface to a Derecho application with standard RPC packages like RESTful RPC or WCF).

What I find interesting is that (1) Derecho can be understood in terms of transactional consistency for atomic "one-shot" operations.  (2) In fact these consistency guarantees extend to aspects of distributed application structure that are not normally considered to need consistency, (3) The really interesting use cases might actually need both Derecho and something like FaRM.

Derecho Primer (mostly duplicates content from the RDMA Derecho posting)

So let's get there step by first.  I'll start with a bit of a duplicate, since I discussed Derecho on the RDMA thread too.  But just in case you skipped that, this illustrates an application of the kind Derecho can help you create:

In this example, the really small white circles are processes: instances of a program, each running on some computer.  The picture thus could span tens or even thousands of nodes in a cluster or datacenter.  We’re trying to help you structure large numbers of processes into cooperating applications where each process plays a specific role within the larger picture.

This particular application has external clients that use browser technologies to talk to it: the red squiggles on the top left.  For example they might use RESTful RPC, which encodes RPC requests as web pages (and similarly for the replies).  REST is widely support but not very fast, which is why we also have a point-to-point RPC layer internal to Derecho.  Most applications using Derecho would use our layer internally even if they talk to the external world using REST or something similar.

The next Derecho concept is that of the top level group.  This is the full membership of the whole application.  In the picture we can see that the top-level group has (at least) 13 members, because there are 13 explicitly shown tiny little white circles.  For most uses the members are actually running identical code, but playing distinct roles.  How can that be?  Well, membership is a kind of input, and each program in a Derecho application knows its respective membership rank (1 of 13, 2 of 13, etc).  So this is enough for them to specialize their behavior.

There is a standard way to specialize behavior, and the picture shows that as well: we associate C++ classes with the top-level group.  This example has 3 classes: LoadBalancer, CacheLayer, and BackEnd.  Each is a normal C++ class, but registered with Derecho.  There is also a notification class, used by the back end to talk to the cache layer (purple circles). 

Derecho uses little functions to define subgroup membership.  They actually are implemented using C++ lambdas.  One such lambda told Derecho that the first three members of the top-level group would be the load-balancer, and we see that at the top.  Another indicated that the cache layer is sharded; in this configuration it seems to have at least 3 shards.  Further, the shards have 3 members each, and there is a shard generator (built into Derecho) that assigned 9 members to this role.  Last, we see that the back end store is also sharded, with 2 replicas per shard, and 4 members in total.

Derecho has all sorts of options for sharding (or at least it will when we finish work on v2 of the system).   Shards can overlap, there can be members held in reserve as warm standby processes to jump in instantly after a crash, etc.  Subgroups and shards can overlap… whatever makes sense to you as the designer.

Next, Derecho has ways to talk to these groups, subgroups or shards: via atomic multicast, which can also be logged into persistent storage.  The latter case matches the specification for Paxos, so in effect, Derecho is a scalable Paxos with an in-memory or a disk execution model, and you can select between the two cases when designing the various components of your application.  The system automatically cleans up after crashes and makes sure that messages are seen in the same order by all members and so forth. 

So with about 15 lines of C++ code per class, more or less, you can instruct Derecho to set up a structure like the one seen here.  And then it just takes a line or so more to send an RPC to some specific member, or issue a multicast to a subgroup or shard, or even to query the members of a subgroup or shard.  The actual delivery of new membership reports (we call them views), multicasts and RPCs is via upcall to methods your class defines; the style of coding looks like a kind of procedure call with polymorophic arguments, and in the case of queries (a multicast to a group where each invoked function returns a result object), you can iterate over the replies in a for loop.

Derecho is blazingly fast.  The API is quite simple and clean, so there isn’t a lot of infrastructure between your code and the RDMA network, and we’ve done some really interesting work to maximize performance of our RDMA sends and multicast operations.  Learn more from the Derecho papers on http://www.cs.cornell.edu/ken, and download the open-source distribution from GitHub.

Derecho plus FaRM 

Visiting at MSRC got me thinking about cases where you might want both Derecho and a transactional DHT, like FaRM or HERD.  I came up with this:

What you see here is two applications, both using Derecho (task A and task B), and then a transactional DHT sitting between them (more properly, present on the same cloud, since all of these span lots of compute nodes).

The case I'm thinking about might arise in a setting like self-driving cars.  Suppose that task A is a helper application for self-driving cars on California highway 101.  It scales out (hence needs lots of read-oriented replicas in the caching layer), has state (car positions and plans), and needs consistency.  And suppose that task B is handling California 280, which is also quite a busy throughway (so it needs its own smart highway helper), but has some overlap with 101.  In fact California might need ten thousand of these tasks just to cover all its busy highway segments.

So we have complex scaled-out applications here, that need replication for high availability, fault-tolerance, and very strong consistency, built using Derecho.  The internal state of each lives within it, including persistency.  But when these tasks share state between tasks -- with external "users", the need is different.  In our illustration, we use a transactional DHT to store state shared between these complex, structured applications.

My thinking here is that the DHT is a very good fit for a kind of anonymous sharing: task A wants to report on vehicles that are likely to leave highway 101 for highway 280, but on the other hand doesn't want to "talk directly" to task B about this, perhaps because the sharing model is simply very convenient this way, or perhaps to avoid interdependencies (we don't want task A jamming up if task B is very busy and slow to respond to direct queries).  Anyhow, the manager for route 1, along the coast, might periodically do a quick scan to plan ahead for cars that might be leaving 101 to join route 1 in the future, etc.  By using the DHT we can support a kind of unplanned sharing.

In contrast the Derecho form of consistency is a great fit for the availability and fault-tolerance needs of our individual tasks -- busy, scaled-out, complex systems, that need consistency.  The DHT wouldn't easily support that sort of application: a DHT like FaRM lacks any notion of a complex, long-lived application with state replicated across its member processes.  FaRM uses replication, but only for its own availability, not to replication application roles or functionality.

So my belief is that we are heading towards a very large-scale world of demanding IoT applications that will really need both models: replication in the form a system like Derecho offers for application design and structure, and transactional DHTs for anonymous sharing.  But I guess we'll have to wait and see whether the application designers faced with building smart highways agree!

Transactions [3]: Replication with consistency (what problem are we solving)?

So the question now arises: what about data replication? 

There are really two issues:

• Data replication inside the professional database, or the DHT, or whatever the transactional server might be.  This is usually completely invisible to the end-user, who just writes little transactional operations and doesn’t need to be aware of how the DHT will handle crashes or similar issues.
• Data replication used in the client systems: either out on the edge (like in a smart car, which might need to have a highly available control system, replicated and continuously active even if some component crashes due to a bug or fails because of a hardware glitch), or in the outer tier of the cloud (e.g. inside the cloud datacenter, but in the layer where the client requests are first received: an example might be an information service launched on behalf of that smart car that will run while the car is actively driving, as its cloud-hosted partner for planning the route, anticipating traffic jams, checking for sales at stores you are passing that sell stuff you seem likely to buy, and so forth).

Both kinds of replication matter, and will often need strong replica consistency, but notice that the replication technology would run in different places:

• The first kind of replication runs inside the transactional service itself, to keep the service alive and accessible even if something fails while it is running.  The earliest work on Lamport’s famous Paxos protocol[1] was something called Viewstamped Replication and was a solution to precisely this problem: Oki and Liskov were building a database and wanted to use replication within it, and ended up with a version of Paxos (it looks just like Lamport’s later Paxos protocol in his famous theoretical treatment a few years later), but deeply integrated with the data storage layer of the database they were building. 
• The second form of replication runs in a highly available application, where we may be replicating the active state of the program (perhaps, data or data structures it uses as it runs) across a set of program instances that back each-other up to step in if a failure occurs.

In my view, this gets to the core of the distinction.  Think back to those edge processes that might be using transactions: in a stateless model, like the one that prevails in the cloud, such a process isn’t fault-tolerant and even if it is playing an important role, like being the representative within the cloud for a self-driving car that connects periodically for updates, the thread doesn’t have any real option for keeping itself alive in the face of routine cloud stuff like elasticity events that shut nodes down (including, perhaps, the node the thread was on).  The cloud does things of that kind all the time, without much consideration of application state, because in the prevailing style of coding any kind of state is supposed to be stored into a database, or a DHT.

This has actually worked very well up to now, but as I see it, the world is shifting because with the growth of Internet of Things applications, multi-user gaming, and other kinds of applications that have continuously managed online state and need to be responsive within a few milliseconds, we can no longer trust the cloud to be fast enough to meet the requirement.  That self-driving car might have a fail-safe approach to handling outages in its cloud-hosted controller, but it won’t want to activate fail-safe mode unnecessarily.  The same is true for smart power grid systems: they can operate as dumb systems, but you often really want high availability.

When you write a distributed program as a set of processes with some form of consistently replicated state, you can also take advantage of having multiple processes to gain extra CPU power, for example to perform actions in parallel.  With modern machine-learning systems, this could matter, so as we move towards a world of active AI and ML applications, that sense real-world input and instantly react, we’ll also move towards a world with greater and greater need for replicated state and consistent coordinated actions by programs that view themselves as team players.  And if that program is long-running, you need it to also tolerate failures, be able to regenerate a node lost due to a crash by integrating a replacement node into the running system, etc.

These needs argue that the future may be a world with far greater use of HPC clusters running applications coded in languages like MPI, and perhaps also far greater deployment of multi-node applications running on general platforms (elsewhere in this blog, on the RDMA discussion, I suggest that HPC with MPI is not a very general infrastructure and that we won’t soon see MPI on datacenters that use fast Ethernet and have multitenancy).  For those cases, we’ll need something different – Cornell’s Derecho system is a response to this specific need.

Back on message, what does this tell us about transactions “versus” consistent replication in Paxos or using atomic multicast (some call this in-memory Paxos or RAM Paxos, but I tend to view it as a different abstraction because Paxos is focused on a stateful model of the protocol itself, whereas atomic multicast is usually finished with a message once it hands it to the application – it isn’t required to log the thing, replay the log later, etc).  Derecho has both.

So in the world I expect will prevail, we’ll probably have a mix of stateless threads on the edge with more stateful, replicated, multi-node services using a replication solution to run in a fault-tolerant way on a set of nodes.  Personally, I think Derecho is the best way to pull this off, replicating state in that kind of a service – much as we did in all the Isis applications years ago.  But if you prefer to use JGroups, LibPaxos, Vsync, RaFT or whatever, I’m not going to try very hard to talk you out of that (basically, a team should use whatever fits best for its goals). 

So our edge would not have these highly available replicated services and applications, side by side with today’s stateless threads.  And my belief is that these highly available, strongly consistent edge services will sometimes need to talk to one-another (service instance to service instance, not internal to the single replicated program).

For example, suppose our highly available service is in charge of helping my self-driving car navigate the New York City traffic and hit the open roads up to Ithaca.  It runs on the cloud, and my self-driving car has an autonomous onboard control system that talks to the service when connectivity is good, getting routing advice and so forth (“Hey Ken, its already 6:30 and we are coming up on a great little restaurant near the Delaware Water Gap: should I reserve a table?”).  And since there might be 1M cars on the road in the US Northeast, there could be 1M instances of this helper running on various US Northeast datacenters.

Where does that transactional DHT or database enter in?  Well, that helper program might be storing my routing plan (like a flight plan) and other information into a database or DHT in order to get help from various backend services that run on big data, or to share my trajectory with other route-helpers for other nearby cars, etc.  Just as the little stateless threads share data via a database or DHT, so would these things.

In contrast, they use consistent replication internally, inside the application, for other purposes, like for high availability, application persistence (e.g. if a node crashes and then restarts), etc.  The benefit compared to storing all the state in the database or DHT is that you get continuous realtime control with the ability to ride out failures or other configuration problems, and you also get an actual programming model for leveraging a set of nodes as part of one service.  This kind of thing is hard to do with a DHT: if your lightweight stateless thread crashes, who relaunches it, and how does it track down its state?  Can it be done in milliseconds or less?  Not obvious.

Think about future banking systems, smart power grid, smart buildings in smart cities, and you get a longer and longer list of possible use cases fitting the pattern.  Which is why I think we need both forms of consistency: replicated services as well as transactional storage layers.

---

[1] For fairness, I should note that this is a much-debated question: the invention of Paxos is one of those events that would normally jet people into contention for a Turing Award, although in this particular case, two of the main contenders already have Turing Awards (for other work, but you can win that award twice). 

Barbara Liskov is quite insistent that she and Brian Oki invented Paxos and she points to the viewstamped replication paper, which appeared in PODC in 1988. 

But my own Isis Toolkit (a data replication tool that I created years earlier than the work Oki and Liskov did) has a protocol called Gbcast in it, for group management, and it bisimulates Paxos, which is a fancy way of saying that any execution of Gbcast matches precisely with an execution of Paxos and vice versa.  So in this mathematical sense, Gbcast really is Paxos, and once one realizes this, the mapping from one to the other becomes more or less evident; there is a Wikipedia article that discusses this, under the heading Gbcast Protocol). 

The trouble is that without a fair amount of manipulation, Gbcast doesn’t look much like Paxos; you need to really think about it to figure out how to transform one into the other, as explained in that Wikipedia article.  So while Gbcast is certainly in the class of protocols that Paxos is in, it isn't clear that one can call it Paxos.  If anything, the opposite seems to be true: Gbcast shouldn't be viewed as a Paxos protocol.

This said, Gbcast definitely solves the same problem that Paxos solves.  Moreover, since we're discussing chronology here, papers on Isis started to appear around 1985, and included this protocol, and I even gave invited talks on the work at MIT in that period.  By 1987 all the main Isis protocols had appeared in major conferences or journals.  But again, the Isis version of Paxos didn’t look much like Paxos, and the proof of correctness was way less elegant than Lamport’s proofs, so even I would have a hard time claiming that this was really the first Paxos.  What I would say is that those who came later would, mostly, have seen this work.

Then, continuing with fair treatment for all, there was yet another protocol, by Larry Stockmeyer, Nancy Lynch and Cynthia Dwork.  The protocol looked more like Paxos than my Gbcast protocol, and had a proof a lot like the one that Lamport later used (not identical), and it was published in the same PODC conference proceedings as the Viewstamped Replication paper, in 1988!  So we have my work in 1985, then these papers which came out simultaneously in 1988, and then Paxos which circulated as a technical report starting around 1990, but didn’t get published until 1996. 

So, who invented Paxos?  I lean towards giving the nod to Barbara and Brian, provided that Gbcast is explicitly acknowledged:

• Gbcast was the first practical protocol to solve the consensus problem in a way that bisimulates what we now think of as the Paxos protocol.  It was not specified in the same manner as Paxos, nor was the proof much like the contemporary proofs.   Let’s say 1985, since that was the date of the first paper on the work, and also the time period where I gave an invited talk at MIT.
• Viewstamped Replication looks like Paxos, and has the famous Synod pattern in it (the core of Paxos), so let’s call this a true Paxos, in 1988.  But the protocol is very deeply integrated with the transactional database they were trying to replicate, and the PODC paper lacked any proofs, so we can’t guess from that at the proof structure.  I’m told that Brian Oki had the proofs in his thesis, but am also told that they didn’t look much like the Paxos proofs.  So: Synod protocol, and a Paxos-style use case, but apparently the proof was not in the style that made Lamport’s Paxos work so famous.
• The consensus protocol of Stockmeyer, Lynch and Dwork, also in 1988.  Another close match, but you need to do some mental mapping to convert it to Paxos (like for Gbcast).  Call it a close runner up.
• True Paxos: let’s date this 1990, when the TR version was widely circulated for the first time.  Has the Synod protocol, the full specification, and the proofs have a modern form.  Definitely the winner if you don’t give full credit to Viewstamped Replication.

As noted, both Lamport and Liskov have already won the Turing Award, but the awards didn’t point to Paxos in either case (Leslie won for his overall body of contributions, starting much earlier with work on causality and time, and Barbara won for early innovations in programming languages and modularity).

Transactions [2]: What about NoSQL?

My quick introduction to transactions may seem harsh to people who love NoSQL, because I more or less brush this approach to the side.  What's my problem?

To get the lay of the land straight, here's a quick explanation of what NoSQL is about.  The idea is mostly associated with key-value storage (also called DHTs, because we hash the keys and implement a distributed hash table): systems that offer a way to do put/get operations with keys of your own design, and values that fit your programming model.  We'll talk about transactional key-value stores separately, but the NoSQL community focuses on non-transactional ones.

Given a DHT but no transactions, what do transactions add to the mix?  The ACID properties are atomicity, consistency, isolation and durability.  Well, if we treat each put or get as a single operation, atomicity comes down to the policy for handling conflicting updates on a single key.  Here, it turns out that cloud systems often solve the problem with a finesse: many don't allow updates and operate in an insert-only model, with automatic garbage collection after a specified delay.  Thus, for many DHTs, there can't be conflicting updates at the level of a single put: end of story!

Even when there could be updates, if our key maps to a specific node and that node does operations one by one, concurrent updates would still be done in some sequentialized order. 

Let's jump past consistency and isolation.  What about durability?  Well, most DHTs do worry about failures, typically by replicating each data item on the node to which the data maps, but also onto the next k nodes along the key-value mapping space.  This way, if k or fewer nodes happen to crash, data won't be lost.  But most commonly, the updates are done by routing the request to the first node in the list of k+1, which then forwards it along the side of the DHT structure (normally, a ring).  So if the first node performs operations in some deterministic order and a small amount of care is taken to forward data in a way that preserves order, the replicas will be consistent with the primary node.

In a nutshell, we've already summarized the NoSQL concept.  Now if you know me at all, you'll know that I really worry when someone expresses a "problem" by describing a "solution".  In some sense, this is what happens with NoSQL: given a key-value system that is almost always able to offer ACID guarantees with no extra effort at all, NoSQL systems often stop there and say "look, this is pretty close to transactional, don't fret about it." 

Historically, the NoSQL movement originated when transactional database systems started to hit serious scalability limits (read the wonderful little paper by Jim Gray, Pat Helland, Patrick O'Neil and Dennis Shasha on this if you haven't yet seen it: they discuss "The Dangers of Database Replication, and a Solution").  Eric Brewer put Jim's point into a very pithy form and proposed a principle he calls CAP, claiming that consistency, availability and partition tolerance are deeply at odds, so much so that you can only get 2 out of the 3.  Eric recommended giving up on consistency: just say no to acid!

And this turns out to work, at least sometimes.  The eBay folks, and Werner Vogels at Amazon, came up with BASE: a methodology for coding with a NoSQL DHT.  Basically, go ahead and design a transactional system, just as you were taught to do at Cornell (if you opted for some other school, well, hopefully they did a good job too...).  But now, get rid of the transaction begin and commit.  Go ahead: just do it!   

Well, without transactions, you'll see a mostly ACID execution for individual put or get operations (mostly, not always), and of course multi-operation sequences can overlap, or be disrupted by a crash.  So the next step is to just pause and think hard about the surprises you might encounter.  The whole point of BASE is that for many web applications, such as shopping on Amazon or picking a movie from Netflix, or bidding on eBay auctions, those surprises won't be so hard to anticipate, or to fix.

So, having thought deeply and identified all the possible oddities... just fix your code to not crash when surprises happen -- instead, just do what you can to hide them from the user.  End of story!  Your system will scale, and nobody will be the wiser about any nasty little inconsistencies that might arise.  In some systems, one also adds a background fixer-upper to check for oddities in the database and repair them (you need to come up with a plan for that too). 

Don't get me wrong: BASE works incredibly well for most web systems, and you can scale a NoSQL DHT pretty much without limits.  So if you are able to follow this route, you will get super performance, amazing scalability, and your users will see the magic sub-100ms response times the web uses as its main design point.  Indeed, since a NoSQL system is potentially inconsistent by design, you can toss caching in anywhere you like, for further speedups.  NoSQL storage systems and the BASE methodology were a home run for the web, without any question.

The issue arises if you work on mission-critical applications with strong safety needs, like medical care systems for use in a hospital, or self-driving cars.  Bankers tend to think money is a life-or-death matter, so they might put financial applications on the list.  Sales people get upset if the customer relationship data is messed up.   So when these kinds of critical computing uses are layered over the NoSQL/BASE model, we have an issue: inconsistency might not be so easy to sweep under the rug.

For example:

• Doctor Smith noticed that patient Sally Jones was having a reaction to a medication, and took her off it.  But the BASE/NoSQL system misfiled the update and now the nursing station just gave her another dose.  Oops!
• The self-driving car you are in just registered its plan to turn left up on Main Street.  But the car planning database dropped that update, and the bus on Main Street believes that the intersection will be completely clear.  Nasty fender-bender...
• The smart power grid just configured itself for an unusually low demand period, but that was a mistake; actually the database system was in a forgetful mood, and neglected to log ten thousand kilowatt hours of demand.  Blackout!

The point being this: not every application can manage with weak consistency.  CAP is fine, for web browsing (the proof being that people use it very successfully, Eric Brewer's company Inktomi was an immense success, and CAP seems like a religion in some communities). But CAP isn't a theorem and in fact there are cases where we need more.

What should we do about this?  One option is to mix ACID with BASE:  you get SALT, my colleague Lorenzo Alvisi's  cool new system.  The basic idea is to study the transactions in isolation, offline, and form groupings: if transactions of flavors A, B and C are all in use at the same time, perhaps we need a costly form of transactional concurrency control, but if we are only running class A transactions, we might be able to manage safely using a NoSQL model, and perhaps when B and C run but not A, there is some small trick that suffices to give full ACID properties, etc.  Lorenzo gets remarkably far this way.  I should have come up with this idea myself!

Another option is to build a transactional DHT.  There are several exciting research demonstrations of this concept by now: FaRM from Microsoft (I visited with the FaRM group for a few months in fall of 2016), HERD from CMU, and there are others too.  For example, my colleague Gun Sirer has an amazing multi-dimensional DHT called Hyperdex that scales incredibly well and has full ACID properties.  In fact all of these solutions work well, and if your application matches their guarantees, they can work so well that you are left puzzled by CAP: is it even more false than we thought?  Could it be that we really can have all three properties, and cheaply?  I think that more and more, the answer is yes!

For me, the bottom line is this: NoSQL and BASE aren't a terrible idea, if your application didn't need ACID in any case.  Eliminating unneeded synchronization delays, without breaking correctness, is a very useful form of optimization.  But the whole point is that if you didn't need ACID, then in some sense, you aren't sacrificing anything when you optimize this way. 

On the other hand, if you actually do need strong properties, you would be a dangerous fool to assume that CAP is a theorem and that the right answer is to sweep inconsistency under the rug.  That self-driving car, or the patient in that ICU -- they might notice, and could be harmed.  NoSQL is fine, for those who don't need the full ACID and SQL model.  But if you do need the model, use one of these transactional DHTs.  They really work, and they scale extremely well too.

Transactions [1]: Transactions in modern systems: A capsule summary

If you are reading this posting, I probably don’t need to tell you what a transaction is, but just to keep our terminology in sync maybe I should start with just a few very standard remarks. 

For me, a transactional data store is any service that resides outside the applications using it, holding state that might be persisted to disk or retained in some form of in-memory representation.  The applications using the service have a way to open a new transactional session (begin a transaction), issue lock, read and write operations, and then can either close the session by making all the updates permanent and releasing locks (commit), or by rolling back all updates (abort). 

In this model we generally think of the application as being stateless: the data lives in the transactional storage systems, while computation on it occurs in these operations that leave no other form of persistent state (obviously, they have variables and can have all sorts of temporary state, but the persistent state is in the transactional database or subsystem, so they are stateless in this very narrow sense, which is what the term really means -- you can throw away all of that local state if these nodes crash).

My definition is at odds with the recent work on transactional memory, but I'll come back to that in a moment.   I'm also not really talking about NoSQL systems, at least not yet.  Those are systems that might offer a transactional API, but that don't provide standard transactional guarantees.  They matter, but they aren't transactional systems, no matter what the vendors may try to claim. 

Among the true transactional systems, there are some important sub-categories.  Quite a few transactional systems provide multi-operation transactions and user-visible locking, so this is the most general case.  Your local SQL database uses this model, and frankly, it is very powerful.  However, at large scale, complex transactions become (potentially) slow.

These days, everyone works with huge databases, so we see more and more systems that have transactions, but rather than try and support transactions when an application writes ten things at once, the model often focuses on real patterns of updates, which very often involve just one thing at a time (and updates themselves are often less than .01% of the traffic).  Since complex transactions are hard to support and the mechanisms introduce high overheads, it is common for transactional big data systems to be limited to so-called one-shot atomic actions (e.g. each operation is atomic, but you don’t get to string them together into multi-operation transactions).  Sometimes people will say that these use linearizability as their model, which is a version of transactional serializability but somewhat less general in its scope, and easier to implement in a scalable fast way. 

The multi-operation and one-shot models are actually more similar than you might expect.  One very cool piece of research a few years back was the Sinfonia system, which showed how you could turn a complicated transactional database into a system that would use one-shot atomic transactions; if you don’t know the work, read about it and the follow-on work at more recent SOSP and OSDIs.  The basic idea was to run a transaction on a snapshot (a read-only consistent copy of the database) and as it runs, track version numbers for any objects it touches.  Then at commit time, you generate a one-shot transaction that just checks to see if these versions are still in use and if so, commits the updates; if not, it aborts and you rerun the transaction using a fresh snapshot.  Very nice idea.

Anyhow, the usual way to describe the full transactional model is by reference to the ACID properties: a transactional system offers atomicity, consistency, isolation and durability. 

Transactional systems exist in many forms; here are the main ones that we work with in cloud settings:

• Database products, which are generally organized around the relational model, and most often support the full multi-operation form of transactions.  Use these if you can, because they are powerful, easy to work with, and don't expose you to surprises (aka "inconsistencies").
• Caches that we use when talking to those databases, which (these days) often use a model called snapshot isolation.  If you find data in the cache it might be stale, but reads will be consistent.  In the cloud people love caches, but be very wary of putting a cache that isn't integrated with the transactional backend store in front of a transactional application.  This is a big trend lately, but can expose you to surprises (see above).
• More and more, we are seeing transactional support associated with large distributed key-value stores |(DHTs): shared memories in which data is associated with a key (which could just be an address, if you want to think of the store as a DSM), and a value (a true DSM would have some fixed number of bytes per address; a DHT has variable sized objects associated with each key).  Not all modern DHTs are transactional, but the transactional ones seem to be far more successful than the flat non-transactional ones.
• Just for completeness, I'll remind you that many people would definitely mention transactional "memory" solutions for concurrent programming in any discussion of transactions.  These typically transform standard objects into transactional objects so that calls to them have transactional semantics.  But I don't really think they belong in the same category. 
• NoSQL systems.  Again, in my way of thinking, these are perfectly interesting systems but simply don't fit in a discussion of the transactional model, because they don't support transactions.  I mention them here mostly because such systems are often marketed as if they could be dropped into place in systems that need full transactions.  In reality, they expose you to surprises (see above).

In this set of blog postings, my focus will turn out to be center on ultra-scalable transactional key-value stores.  These are the new guy on the street (relatively speaking) and because they scale so incredibly well, have the potential to be a big deal, once they mature.  But before we go there, I have a few remarks about the other models too.

A first question centers on whether we are thinking about the transaction system as the whole story (more or less, PaaS), or as a tool (as in the case of a DHT).

One thing to be aware of is that when you use a professional database, like Oracle’s enterprise solution or SQL server, the database has the whole story for data lifecycle management, built right in.  For example, a bank needs to encrypt customer data, and must keep it for N years, but then must delete it.  And it needs to ensure that only the proper people have access, that accesses are audited, that security alarms go off if someone tries to steal the money, etc.  A professional database has solutions for all of this, is designed for massive scaling and optimized for high speed.

The transaction DHT solutions are a new entrant to the market and are much less mature.  These systems can hold a lot of data, and they do offer transactions, but they would often lack cryptographic tools, won’t have services for long-term archival data management, etc.  For example, if a runaway (buggy) application dumped bad records into a professional SQL system, you could easily identify them and remove them later.  But a DHT might just bloat silently with no simple way to even list the data it contains.  You could end up with gigabytes of orphaned data: the system will hold it against further need, yet the application never intended to generate it, and in fact will never access it again.  Over time this will change as the DHTs catch up (for example, by supporting debugging development mode, data curation tools, automatic deletion after a lease expires, etc.), but it will take a while.

Then there is a question of what the application using the transaction system really looks like. 

In-memory transactions,  So let me return to this now and say a few words about the idea, and then set it to the side.  In-memory transactions were a hot topic within the programming languages community, and were explored very actively starting a few years ago.  The basic idea is simple: because so many programmers have trouble with concurrency, the inventors of the approach wanted to automatically insert the needed concurrency control so that operations on objects in a standard programming language like C++ or Java could be performed as small transactions.  Since the compiler would do this in a provably correct way, the bug rate experienced by developers would drop sharply.  At least that was the idea.

Without discussing the concept at length, I'll just summarize some of the issues that surfaced and ultimately seem to have limited adoption of that approach.  In fact none of these issues are at all new: Barbara Liskov proposed transactions on objects way back in the 1980's (the Argus project).  Argus got quite far, but on the other hand, ran into almost all the same questions.  These include:

1. Performance issues.  The locking or versioning schemes required were more costly than the developers had expected, hence developers balked at using them.  For example, without being clever, many transactional object oriented systems risk holding locks for very long periods of time, because operations block for I/O or while waiting for other forms of input.  This can cause other concurrent threads to end up waiting, and can even introduce deadlocks.  A clever programmer would be able to avoid such issues, but asking the compiler or runtime system to do so is asking a lot.  Thus the developer either find the transactional solution buggy and slow, or will anticipate trouble and may try to break the transactional model in order to evade its limitations (which causes other issues).
2. Instability.  Rachid Guerraoui (who actually likes software transactions a lot) points out that potentially, a transactional object could get into a state where two or more competing transactions basically livelock by causing one-another-to abort and retry.  You might expect this to be hard to trigger, but it turns out that in object oriented programming, you can easily trigger such scenarios, particularly if developers try and "break" the model to gain performance.  So you start with an object oriented system that was safe and live (would guarantee progress), were you to use locking carefully to avoid concurrency bugs, and then by reexpressing it using transactional objects, it suddenly starts to have infinite loops (livelock).
3. Limitations of the model.  This last point touches on a basic issue: if thread A is in a transaction and wants to pass information to thread B, technically speaking that isn't permitted until the transaction commits.  But developers find this frustrating and may try and work around the issue by performing actions with external side-effects in ways that the transactional system can't easily control, such as I/O.
4. Nested transactions.  There is a lot of debate around the proper way to handle recursion and other forms of nested calls.  Argus had a neat approach called nested transactions, but it can have very high overheads and requires pretty complex supporting mechanisms.  But without nested transactions, the transactional programming model might not allow recursion or other forms of nested operations: quite a serious restriction, since modern languages treat almost any operation as a method invocation!
5. I/O is common, but transaction systems shouldn't try to do write operations from within an uncommitted transaction: if they do, the model requires that the data be held back until the transaction commits.  Similarly, a read may need to push data back on the input channel if the transaction that did the read happens to abort.  Both limitations are costly and may be impractical.
6. Multicore parallelism isn't necessarily ideally suited for multi-threaded programming, in any case.  Multicore machines are awesome for supporting multitenancy: you launch hundreds of VMs, and each core owns a few of them, but the programs in those VMs are standard ones with relatively few threads.  To write multithreaded code that will actually speed up with multiple cores is a delicate art: you need to understand the NUMA memory behavior of your system and the layout of data into memory, and think hard about locking and other forms of concurrency control, you may need to worry about priority inversions or other oddities of scheduling, etc.  So the whole premise of the transactional memory work centers on assuming that this kind of programming is very valuable, and that hasn't really been shown to be the case.

All of these reasons lead me to view transactions on memory objects as a different area of research, with its own goals, its own user community, and its own challenges.

Lets pivot back to our main topic and focus on transactional ways of interacting with external data. 

Now the first thing to appreciate is that transactional access to external data is incredibly valuable and for this reason, incredibly popular.   Technologies like Microsoft’s .NET and Oracle’s versions of Java let your application map transactional subsystems into the application address space: the data appears to be in a set of collections (the in-memory name for a key-value store) and you access it through a programming language API that mimics the database SQL language, such as the .NET LINQ (Language Integrated Queries) API, or the Java equivalent, called JQuery.

Let's focus purely on this second category of transactional systems and ask how well each fits with the modern cloud.

Stateless applications that are part of some form of cloud platform (often a PaaS solution such as Microsoft Azure or Oracle’s three-tier cloud platform, IBM’s WebSphere, etc) often benefit from transactional mechanisms of these kinds.  The "client" systems are generally web browsers that issue requests.  Each request routes into a server in the cloud, where a thread executes on its behalf. 

That thread has the data the browser client sent in, as well as full access to the cloud, including transactional caches and back-end servers, and it can use files on the local file system.  Nonetheless, when the thread terminates, any data it saved locally in files will be deleted (so stateless really means automatically garbage collected).  Saved state is always persisted into the database or the transactional storage layer.   Most cloud applications work this way.

So with this background, we can talk about some typical modern cloud applications that might use transactions.  The standard example is something like this:

• A banking application permits you to set up a wire transfer to pay bills.   You fill in the transfer page and click “schedule payment”.
• The banking application generates a call to a web services method running on a stateless cloud node (in the sense mentioned above), and a thread launches to receive and process the request.
• The schedule-payment method runs a transaction to read your account balance, check that the request is legal, and then either commits the transaction to the pending payments ledger or aborts and reports an error, such as “account overdrawn”.

So what’s not to love?  I’ll come back to that in a minute rather than explain here, but the short answer is that transactions are totally awesome, at least for uses like this, but at massive scale they do bring a lot of issues too.  But this blog posting is really long by now, so I'll pause here.  In the other blog sub-postings, I touch on them one by one.

Transactions: [Topic header] Using transactions side-by-side with group replication

Transactions and strongly consistent group replication solve different problems.  We need both.

A quick summary of this whole set of postings:

·         Transactions, and in particular, transactional key-value stores, are a genuinely exciting technology for sharing information between tasks that are mutually anonymous.  By this, I mean tasks A and B that update and share information about some sort of external reality, but where A has no direct knowledge of B and vice versa: they are not directly coordinating their actions, or running as two parts of some overarching structured application.

o   The ACID properties make sense for this use case.

o   The model of lightweight tasks that use the transactional memory as their sole persistent store is very natural for the cloud, so we see this approach in PaaS solutions.

o   Often we say that the tasks are stateless, meaning that they always are launched in whatever the original state defined by the virtual machine or the code is; state is not kept from activation to activation (except in the transactional store, where state persists).

·         Group replication is a great model for high availability programming.  You design with a state machine style of execution in mind, although generally the rigid lock-step behavior of a state machine occurs only in replicated objects that might be just a small subset of the functionality of the overall application. 

o   For example, a modern application might have a load-balancer, a cache that is further subdivided into shards using a key-value model, and then each of the shards might be considered as a state machine replicated group for updates, but its members might operate independently for RPC-style read-only operations (to get read parallelism).   Then perhaps the same application might even have a stateful persistent backend.

o   We use virtually synchronous atomic multicast and virtually synchronous Paxos in such applications: virtual synchrony ensures consistency of the membership updates, and then the atomic multicast or Paxos protocol handles delivery of messages or state updates within the group.

·         In today’s cloud we might want both functionalities, side by side. 

o   The virtual synchrony group mechanisms make it easy to build a scalable fault-tolerant distributed program that will be highly available, responding very rapidly to external stimuli.  You would want such a program in an Internet of Things setting, to give one example.

o   But if you run such a replicated program many times, separately, the program as a whole is quite similar to the tasks mentioned in the first bullet: each of these multi-component programs is like a task.  And we might do that for a setting like a cloud-hosted controller for a smart car, where we want high availability and speed (hence the group structure), but also prefer to have one of these applications (replication and all) per car.  So we could launch a half million of them on the same cloud, if we were interested in controlling the half million or so cars driving on the California state highway network on a typical afternoon.

o   These group-structured programs might then use a transactional key-value store to share data between instances (in effect, the car-to-car dialog would be through a shared transactional store, while the single-car-control aspect would be within the group).

Here’s a picture illustrating a group-structured program like the one just mentioned:

In this example we actually see even more groups than were mentioned above: the back-end is sharded too, and then there are groups shown that might be used by the back-end to notify the cache layer when updates occur, as is needed in a so-called coherently cached, strongly consistent store.  Paxos would be useful in the back-end, whereas the cache layer would be stateless and hence only needs atomic multicast (which is much faster).  Applications like this are easy to build with Derecho, and they run at the full speed of your RDMA network, if you have RDMA deployed.

The point then is that replication is used here to ensure rapid failure recovery of the application instance (this is the data in the back-end store), and for scaling to handle high query rates (the replication in the cache layer).  But this form of replication is entirely focused on local state: the information needed to control one car, for example.

I tried to make a picture with a few of these structured applications sharing information through a key-value store, but it gets a little complex if I show more than two.  So here are two of them: task A on the left and task B on the right.  The transactional DHT in the middle could be FaRM or Herd:

Notice the distinction: local state was replicated in each of task A and task B (handling two different vehicles), but the global state that defines the full state of the world, as it were, is stored in the transactional key-value database that they share.  This matters: processes running within A or within B know of the structure of the overall car-manager task in which they run.  Subtasks of a single task actually know about one-another and cooperate actively.  In contrast, when we consider task A as a unit, and task B as a unit, the picture changes:  A does not really know that B is active and they do not normally cooperate directly.  Instead, they share data through an the DHT: the transactional key-value subsystem.

As in the case of RDMA, there is a lot more to be said here.  I've posted sub-topic remarks on a few of the points that I find most interesting.