Parallel Commits: A New Atomic Commit Protocol for Distributed Transactions

https://www.cockroachlabs.com/blog/parallel-commits

by dilloc • 7 years ago

184 38 7 years ago

aboodman4• 7 years ago

Why is it so important for transactions to be interactive? I feel like this is not a feature that I use frequently as an application developer and it makes distributed transactions so much harder.

It seems like something that accidentally came along from SQL, and is imposing a large cost.

zaphar• 7 years ago

Interactive may be a little misleading here. Consider a common case in any java application at my current place of employment.

1. A controller starts a transaction.

2. Some code looks up a record in a database.

3. Based on the results of that lookup we run 1 of two possible writes to a table in that database.

Knowing what kind of write you will do in the transaction requires knowing the result of the lookup which itself must be run in the transaction. Now take this small case and imagine it in the real world scenario where there might many such lookup specific write logic all wrapped in a single transaction and the space of possible combinations of statements you will want to execute in the transaction is large. None of this is interactive in the sense of someone working in a shell. But it is interactive in a way that makes the transaction useful in the real word.

aboodman4• 7 years ago

Exactly. What Calvin (and FaunaDB, I think) does is allow you to run an arbitrary function which takes input on the database. That function can do whatever it wants, including any number of reads, writes, and arbitrary logic based on the reads. But critically, it can't talk back to the calling client, except for to send the final result.

This allows you to implement the pattern you describe, which I agree is common, but in a dramatically simpler way.

Having the database, which is not really a single thing, but a swarm of computers spread across the globe separated by unreliable links who are trying to stay in consensus, pause their work to hear back from the client just seems ... well it seems miraculous that it can work at all.

The Calvin way is so much easier, it seems like there must be some very good reason that it's not what CockroachDB does. But I've never heard what that reason is.

BubRoss• 7 years ago

This article implies that it reduces latency.

cryptica3• 7 years ago

I like the elegance and simplicity of two-phase commits. I didn't understand the criticism in the article; maybe it's something specific to CockroachDB. In my experience with two-phase commits, if the system crashes before a transaction is fully processed and committed, it should be fully reprocessed (from scratch) once the server restarts.

In one of my open source projects with RethinkDB (which doesn't natively support atomic transactions), I implemented a distributed 'parallel' two-phase commit mechanism by assigning each pending record a shard key (an integer derived from a hash of the record id); then worker servers would decide which subset/range of DB records to process/commit based on a hash of their own server id (which would tell them which range of shard keys/records they were responsible for). Only when a record had been fully processed, its status would be updated as committed by writing a 'settled' flag on that record.

Whenever a server failed and restarted, it would pick up processing from the last successful commit. If a server did not restart, the worker count would be updated and remaining workers would redistribute the partitions among themselves based on the shard keys of the records.

irfansharif7• 7 years ago

The "criticism" as it pertains to CockroachDB and off-the-shelf 2PC is less so about 2PC in isolation and more so about layering 2PC on top of consensus groups used to persist records. When any given txn does the "prepare" phase, it lays down markers for a possible upcoming commit. If the 2PC coordinator fails in an inopportune moment, there's a delay between the failure and the markers being cleaned up (whether or not the transaction is "reprocessed" or aborted). The reason why this delay is problematic is because any subsequent transactions that happen upon said markers, they just have to wait for the resolution ("commit"/"aborted") aka it blocks. So clearly recovery must be built into 2PC, i.e. the transaction state itself must be persisted. This is done so in the same way the markers/regular writes are, through consensus. But marking the transaction state as "committed" can only happen once we're guaranteed that all the individual write markers are persisted. Which adds a second round of consensus.

ryanworl• 7 years ago

This is about reducing the number of message delays before the commit succeeds. Failure scenarios have to be handled to be correct, but this is a performance optimization primarily from what I can see.

mjb• 7 years ago

Whenever a server failed and restarted, it would pick up processing from the last successful commit. If a server did not restart, the worker count would be updated and remaining workers would redistribute the partitions among themselves based on the shard keys of the records.

You don't describe how you come to a global view of which is the "last successful commit". Do you mean that the co-ordinator recovers (such as from its own log) and uses that information, or do you mean you use local information? In the latter case, what you describe doesn't sound like it preserves transaction atomicity across node crashes.

cryptica3• 7 years ago

In RethinkDB, a table can be split into shards and each shard has at most one master and multiple replicas. With one master to perform all writes for a shard, there is no need for a co-ordinator. If the master of a shard goes down, a replica can be promoted as the master. For writes, RethinkDB prioritizes consistency over availability so a server failure may carry some downtime in terms of writes but reads can be configured to be highly available (though they could potentially be out of date).

But there is definitely a downside that if a RethinkDB master server fails immediately after a write (before it propagated to a replica) and does not recover, then there can be some data loss (but only the most recent writes on that shard).

jlokier3• 7 years ago

In my experience with two-phase commits, if the system crashes before a transaction is fully processed and committed, it should be fully reprocessed (from scratch) once the server restarts.

Then the problem becomes, in a fully distributed system how do you know which was the last successful commit before the system crashed?

That 'settled' flag you mention may be set in the coordinator node's storage just before the coordinator crashed, but not communicated to any other node because of the crash.

So the other nodes have to wait for the coordinator to restart, and replay its storage, to find out if that transaction was successful.

That pause can be quite long, especially if it involves a coordinator rebooting, and really long if it involves resynchronising a RAID, checking filesystems/DB integrity, etc.

The other nodes could decide it doesn't matter, exclude the coordinator from the cluster, and reprocess starting from an older transaction. But then they have to agree which older transaction - a distributed consensus problem.

This is certainly possible, but I wouldn't call it simple, and it wouldn't be two-phase commit any more.

Restarting older transactions tends to be visible to clients as well, because they may see transaction retries, and it's often desirable to minimise the number of those from a distributed database for minor failures (such as one node harmlessly going offline). Among other things, they can cause load spikes in the large system, and may exercise retry corner cases that don't come up normally, surfacing bugs that shouldn't be there in application code, but are.

foota2• 7 years ago

Wow, this is impressive.

cdbattags• 7 years ago

What I'll say might come off naive but I accept this at an attempt for a reductionist's viewpoint.

A singular update that side affects *N (times N) where N is greater than 1 will always lead to either a race condition or latency.

ComodoHacker4• 7 years ago

From what I understand, with this new commit protocol they managed to improve response time of a writing transaction by shifting some work of determining its final status to the readers. Am I understanding correctly that readers' performance will degrade by the same amount?

While this is an achievement which some scenarios will definitely benefit from, like bulk loading or updating secondary indexes as mentioned in the article, what about other scenarios where read performance is more important, like the ones where there are much more readers than writers? Shouldn't there be a configuration option which commit protocol to use?

irfansharif7• 7 years ago

Am I understanding correctly that readers' performance will degrade by the same amount?

Not quite. The "slow path" talked about is only applicable when the coordinator node is unavailable (presumably a rare event). If it's unavailable, there's nobody left to clean up the STAGING txn record, so the reader is tasked to do it itself.

In normal conditions however, once the coordinator node receives acknowledgement for the successful persisting of all write intents and the "txn STAGING" record, it can simply record "txn COMMITTED" in memory (and return to the client, send off async intent resolution procedures, etc.) Any subsequent read requests that observes left over intents (yet to be resolved) are pointed to the coordinator node, which can simply consult the "txn COMMITTED" record in memory. This is all safe because the commit marker is not simply stored on the coordinator node, it's a distributed condition and can be reconstructed by any observer even if the coordinator failed.

ComodoHacker4• 7 years ago

Any subsequent read requests that observes left over intents (yet to be resolved) are pointed to the coordinator node, which can simply consult the "txn COMMITTED" record in memory.

Another roundtrip performed by reader rather than writer? That's what I'm talking about.

Though I understand it differently, as reader just waits until all writes and "txn COMMITTED" record arrive at it's node.

grogers• 7 years ago

Given how critical preventing future intent writes is to the protocol to ensure safety during recovery, it'd be nice to have more detail on how that works. Calling it an in memory data structure doesn't exactly inspire confidence.

irfansharif7• 7 years ago

Agreed, we should be talking about the "timestamp cache" in more detail generally. While I'm here, looking at [1] helped me confirm how everything is kosher despite being in-memory. The timestamp cache pessimistically maintains a "low water mark", this always ratchets up monotonically and represents the earliest access timestamp of any key in the range. Writes happening at timestamps lower than this water mark are not let through, and bumping this watermark past the point of the observed write intent's timestamp is how slow inflight write intents are aborted by recovering read requests. On server restart, the timestamp cache is initialized with a low water mark of the current system time + maximum clock offset, so "future" write intents (or more accurately: write intents sent in the past but previously stuck in transit) are simply rejected.

[1] https://github.com/cockroachdb/cockroach/blob/master/pkg/sto...

mcms• 7 years ago

What prevents a transaction from being prematurely aborted while there are some intents in flight?

From what I understand, transaction A can still be replicating intents and transaction B, not knowing that coordinator for transaction A is still at work, start a recovery process. This makes transaction A abort which could be prevented by waiting for intents of A to be successfully replicated.

Parallel Commits: A New Atomic Commit Protocol for Distributed Transactions

Related Stories