6 ms·

Non-blocking transactional atomicity

136 points by sethev 13y ago

antirez 13y ago
Very cool and simple. The notion of "time" may be a bit misleading but actually the client just requires to generate an unguessable rand, think at it like getting a string from /dev/urandom, so actually time is completely not part of the protocol which is great (also avoiding to deal with client IDs can be nice in practice).
- pbailis 13y ago
  Yep--thanks! I've updated the post accordingly :)
ch 13y ago
In the mode where a client is responsible to move writes from 'pending' to 'good', it wasn't clear to me what happens if the client dies before contacting all servers in the second pass? Does the data remain stable, or must some additional work be performed to correct the inconsistent state?
- pbailis 13y ago
  > Does the data remain stable, or must some additional work be performed to correct the inconsistent state? If you want client writes that reached all servers to become visible, then the servers will have to perform the move from 'pending' to 'good' on their own (by communicating asynchronously). The notification of write stability is idempotent, so it doesn't hurt if both clients and servers perform this notification. FWIW, in our implementation, servers perform the second step instead of clients (which can be made more efficient via batching).
leef 13y ago
The algorithm does seem nicer than a 2PC protocol in that there is no need for a 2PC coordinator. By distributing the metadata as it does the clients and servers can figure out what has been committed and what hasn't. However it doesn't appear to directly include semantics for aborting transactions which is a pretty important part of a distributed transaction protocol. The paper admits the algorithm does not guarantee termination but I would have liked to see more details on the failure scenarios regardless (minor details in footnote 3). It's not clear to me what writers see (if anything) when a write fails. The paper does talk about how non-overlapping transactions won't block each other (which is nice but not a solution) and how one could add the ability to abort and trigger a cleanup by the use of a good (user supplied) failure detection module. But having a reliable node failure monitor that can react fast enough to ensure availability is really the hard part. Would love to see more on aborting transactions next.
- pbailis 13y ago
  > However it doesn't appear to directly include semantics for aborting transactions which is a pretty important part of a distributed transaction protocol. Yep, I left this out to avoid confusion at first. There are some details in the "What just happened?", but the basic idea is that any aborted write will be stuck in "pending." Same for failed writes; writers won't see these. The algorithm presented actually guarantees "Read Committed" ACID isolation. > But having a reliable node failure monitor that can react fast enough to ensure availability is really the hard part. Well, you'll remain available for reads and writes, but the size of "pending" might grow. You essentially need asynchronous distributed garbage collection, which will stall in the presence of partitions and may require the failure detectors I mentioned. > The paper does talk about how non-overlapping transactions won't block each other (which is nice but not a solution) I don't see how this isn't a solution for transactions that desire last-writer-wins semantics. If, as in the examples I listed, writes commute, then a blocked write shouldn't stall others. If you want to prevent Lost Update or Write Skew anomalies (i.e., concurrent update), then you'll have to give up availability and/or block.
  - IgorPartola 13y ago
    > I don't see how this isn't a solution for transactions that desire last-writer-wins semantics. If, as in the examples I listed, writes commute, then a blocked write shouldn't stall others. Doesn't writes that commute mean that there was no contention to begin with? Ideally, if I have a balance of $100 in my account and try to spend $60 in two different transactions, one should come back as failed before the purchases are complete. > If you want to prevent Lost Update or Write Skew anomalies (i.e., concurrent update), then you'll have to give up availability and/or block. There is a difference between giving up read and write availability. My ideal database should be read-available at all times, but guarantee that writes are atomic and durable (and give up availability for this guarantee). On the whole, this looks pretty neat. I like the idea of the client being responsible for the writes being committed on the servers. The client is then free to choose how to implement the IO, but ultimately, if a single client experiences a failure and a single write doesn't go through, it is usually a better outcome than a write going through and then replication between two servers breaking. What are your thoughts on quorum-based voting in distributed systems? E.g.: your protocol but with the requirement that a write is considered stable if only most (vs all) of the servers involved have it marked as "good".
jeffcutsinger 13y ago
If you clear metadata, the following pathological case results in an inconsistent read: * x = 0 and y = 0, starting out in the "good" state with metadata cleared; * client A reads x with no metadata, then for whatever reason blocks or is delayed; * client B writes x = 1 and y = 1, completes the transaction, and the metadata is cleared; * client A reads y = 1. This shouldn't be an issue unless you aggressively clear metadata and have a long-running client.
- IgorPartola 13y ago
  In that case, if client A wants to do something with the value x, it should instruct the server to place metadata on it, essentially saying that it's reading it with the intent to update. This could be implemented as "update x with value of get(x)". That way it would be placed into pending with metadata. Having a heartbeat between the client and the server could also help prevent the write from getting purged from "pending" prematurely.
  - pbailis 13y ago
    I'll also add that the metadata requirements aren't huge--typically 8 bytes for the timestamp and N*(bytes per key) for the keys. In our implementation and the benchmarks that we provided, we don't garbage collect the metadata as it's small and isn't a serious overhead.
ww520 13y ago
This is very cool. Simple enough to reason the invariants easily. I guess one of the key insights is that each data has a canonical server owner which enforces the consistency of the writes of the data at a single place. I have one question regarding the determination of the latest version of a set of peer data when overlapping transactions occurred. Suppose initially s1/x=0 at server s1 and s2/y=0 at server s2. Client1 updates s1/x=10 and s2/y=10 at transaction v1, and at the same time client2 updates s1/x=20 and s2/y=20 at v2. Suppose the clients contact the servers in different order and the update messages arrive at the servers in reverse order, such that s1's pending queue is [x.v1=10, x.v2=20] and s2's pending queue is [y.v1=20, y.v2=10]. After client1 and client2 send the make-good command, s1's good queue is [x.v1=10, x.v2=20] and s2's good queue is [y.v2=20, y.v1=10]. When a third client3 tries to read the latest value of x or y, what is the latest value of its peer data? It looks like depending which data client3 starts with, it would get a different version of the peer data? Like starts with s1/x got s1/x.v2 == 20 and s2/y.v2 == 20, while starts with s2/y.v1 == 10 and s1/x.v1 == 10. Am I missing something or this is the semantic in determining the latest values of peer data?
- pbailis 13y ago
  (edit: post author, [not OP]) here. Thanks for the feedback! > I guess one of the key insights is that each data has a canonical server owner which enforces the consistency of the writes of the data at a single place. Well, this is the way I presented it, because it's easiest to understand. But, if you want a replicated system that provides HA, there are alternatives (http://www.bailis.org/blog/non-blocking-transactional-atomicity/#ha-note http://www.bailis.org/blog/non-blocking-transactional-atomic...). > When a third client3 tries to read the latest value of x or y, what is the latest value of its peer data? It looks like depending which data client3 starts with, it would get a different version of the peer data? ... Am I missing something or this is the semantic in determining the latest values of peer data? Good question! This ultimately comes down to how you want to handle concurrent writes. Many distributed databases use a "last writer wins" strategy when reconciling concurrent updates (that is, the correct behavior is specified to be that the database serves the highest-timestamped version of a given data item). Now, in your example, the clients both started their writes at the same (real-world) time and used this time as a basis for their timestamp, so the "last" write is undefined. We need a way to break the "last writer wins" tie. In practice, this can be something like the client ID appended to the last few bits of the timestamp or even a hash of the value written--as long as the tie-breaker is deterministic (that is, different replicas don't decide different "winners"), it doesn't really matter which is chosen. In practice, to avoid storing every value ever written, you'd want to provide some kind of "merge" function for multiple writes, and, in our implementation and in often practice, this is last writer wins (plus some deterministic tie-breaker for identically timestamped but distinct writes).
dscrd 13y ago
I think there's a typo under "good, pending and Invariants"? Isn't the caption supposed to be "not okay / (x=1 should be in pending or good)"
- pbailis 13y ago
  Post author here. Hmm. The invariant we're trying to maintain is that any write in 'good' should have its transactional "siblings" in either 'good' or 'pending' on their respective servers. So if we are trying to write x=1 and y=1, then, if x=1 is in the x server's 'good', then y=1 should be in the y server's 'good' or 'pending'. But in the example under "not okay", y is not present in either.
  - dscrd 13y ago
    Ok, then I just did not understand that part yet.
yxhuvud 13y ago
Getting only 4.8% percent of throughput for reads doesn't seem stellar. Why is this so bad?
- pbailis 13y ago
  Ha--good catch! I don't know if anyone else noticed that. I meant to say "within 33-4.8% of the peak throughput..." That is, we're 95.2% of the peak, not 4.8% of the peak. Fixed, thanks!
espeed 13y ago
This model is somewhat similar to how Datomic models transactions -- Datomic uses immutable datoms and includes a transaction ID with each datom: "A datom consists of an entity, attribute, value and transaction (time)" (http://www.datomic.com/rationale.html http://www.datomic.com/rationale.html). However, Datomic's design is based on a single, centralized transactor that pushes out all the new transaction information to an index that is distributed to all its clients (peers), whereas NBTA would enable distributed transactors. See "The Datomic Data Model" http://www.datomic.com/rationale.html#DataModel http://www.datomic.com/rationale.html#DataModel From Datomic's FAQ http://www.datomic.com/faq.html http://www.datomic.com/faq.html ... How does Datomic provide ACID guarantees? The transactor serializes all transactions, and each transaction runs against a stable view of the database, and succeed or fail in their entirety. Transactions are not acknowledged until after they are logged in storage. Atomic read/modify/write operations can be performed by database functions, which run on the transactor within the transaction. Note that Datomic can provide ACID guarantees without utilizing read-transactions, nor read locks, due to the presentation to the query engine(s) of the database as an immutable value. ---//--- Rich Hickey on Datomic's design... Intro to Datomic (12 min) http://www.youtube.com/watch?v=RKcqYZZ9RDY# http://www.youtube.com/watch?v=RKcqYZZ9RDY#! The Design of Datomic (60 min) http://www.infoq.com/presentations/The-Design-of-Datomic http://www.infoq.com/presentations/The-Design-of-Datomic The Datomic Architecture and Data Model (46 min) https://vimeo.com/45136212 https://vimeo.com/45136212
- pbailis 13y ago
  Very cool, thanks! In general, immutable data items makes NBTA much easier; there's only one element in either 'good' or 'pending' for each data item at a time. The benefit of a single centralized transactor/ID generator is that it provides a simple total ordering on updates. The downside is that updates must (often) complete sequentially to guarantee safety and that the centralized service can become a bottleneck/challenge during failures. That said, I'm a big fan of the Datomic vision and look forward to seeing what they're able to accomplish.
agilord 13y ago
In theory this idea looks OK. I'd be interested if someone implemented it on Riak or Cassandra? What are the conclusions?

136 points by sethev 13y ago

Very cool and simple. The notion of "time" may be a bit misleading but actually the client just requires to generate an unguessable rand, think at it like getting a string from /dev/urandom, so actually time is completely not part of the protocol which is great (also avoiding to deal with client IDs can be nice in practice).

pbailis 13y ago
Yep--thanks! I've updated the post accordingly :)

ch 13y ago

In the mode where a client is responsible to move writes from 'pending' to 'good', it wasn't clear to me what happens if the client dies before contacting all servers in the second pass? Does the data remain stable, or must some additional work be performed to correct the inconsistent state?

pbailis 13y ago
> Does the data remain stable, or must some additional work be performed to correct the inconsistent state? If you want client writes that reached all servers to become visible, then the servers will have to perform the move from 'pending' to 'good' on their own (by communicating asynchronously). The notification of write stability is idempotent, so it doesn't hurt if both clients and servers perform this notification. FWIW, in our implementation, servers perform the second step instead of clients (which can be made more efficient via batching).

leef 13y ago

The algorithm does seem nicer than a 2PC protocol in that there is no need for a 2PC coordinator. By distributing the metadata as it does the clients and servers can figure out what has been committed and what hasn't. However it doesn't appear to directly include semantics for aborting transactions which is a pretty important part of a distributed transaction protocol. The paper admits the algorithm does not guarantee termination but I would have liked to see more details on the failure scenarios regardless (minor details in footnote 3). It's not clear to me what writers see (if anything) when a write fails. The paper does talk about how non-overlapping transactions won't block each other (which is nice but not a solution) and how one could add the ability to abort and trigger a cleanup by the use of a good (user supplied) failure detection module. But having a reliable node failure monitor that can react fast enough to ensure availability is really the hard part. Would love to see more on aborting transactions next.

pbailis 13y ago
> However it doesn't appear to directly include semantics for aborting transactions which is a pretty important part of a distributed transaction protocol. Yep, I left this out to avoid confusion at first. There are some details in the "What just happened?", but the basic idea is that any aborted write will be stuck in "pending." Same for failed writes; writers won't see these. The algorithm presented actually guarantees "Read Committed" ACID isolation. > But having a reliable node failure monitor that can react fast enough to ensure availability is really the hard part. Well, you'll remain available for reads and writes, but the size of "pending" might grow. You essentially need asynchronous distributed garbage collection, which will stall in the presence of partitions and may require the failure detectors I mentioned. > The paper does talk about how non-overlapping transactions won't block each other (which is nice but not a solution) I don't see how this isn't a solution for transactions that desire last-writer-wins semantics. If, as in the examples I listed, writes commute, then a blocked write shouldn't stall others. If you want to prevent Lost Update or Write Skew anomalies (i.e., concurrent update), then you'll have to give up availability and/or block.
- IgorPartola 13y ago
  > I don't see how this isn't a solution for transactions that desire last-writer-wins semantics. If, as in the examples I listed, writes commute, then a blocked write shouldn't stall others. Doesn't writes that commute mean that there was no contention to begin with? Ideally, if I have a balance of $100 in my account and try to spend $60 in two different transactions, one should come back as failed before the purchases are complete. > If you want to prevent Lost Update or Write Skew anomalies (i.e., concurrent update), then you'll have to give up availability and/or block. There is a difference between giving up read and write availability. My ideal database should be read-available at all times, but guarantee that writes are atomic and durable (and give up availability for this guarantee). On the whole, this looks pretty neat. I like the idea of the client being responsible for the writes being committed on the servers. The client is then free to choose how to implement the IO, but ultimately, if a single client experiences a failure and a single write doesn't go through, it is usually a better outcome than a write going through and then replication between two servers breaking. What are your thoughts on quorum-based voting in distributed systems? E.g.: your protocol but with the requirement that a write is considered stable if only most (vs all) of the servers involved have it marked as "good".

jeffcutsinger 13y ago

If you clear metadata, the following pathological case results in an inconsistent read: * x = 0 and y = 0, starting out in the "good" state with metadata cleared; * client A reads x with no metadata, then for whatever reason blocks or is delayed; * client B writes x = 1 and y = 1, completes the transaction, and the metadata is cleared; * client A reads y = 1. This shouldn't be an issue unless you aggressively clear metadata and have a long-running client.

IgorPartola 13y ago
In that case, if client A wants to do something with the value x, it should instruct the server to place metadata on it, essentially saying that it's reading it with the intent to update. This could be implemented as "update x with value of get(x)". That way it would be placed into pending with metadata. Having a heartbeat between the client and the server could also help prevent the write from getting purged from "pending" prematurely.
- pbailis 13y ago
  I'll also add that the metadata requirements aren't huge--typically 8 bytes for the timestamp and N*(bytes per key) for the keys. In our implementation and the benchmarks that we provided, we don't garbage collect the metadata as it's small and isn't a serious overhead.

ww520 13y ago

This is very cool. Simple enough to reason the invariants easily. I guess one of the key insights is that each data has a canonical server owner which enforces the consistency of the writes of the data at a single place. I have one question regarding the determination of the latest version of a set of peer data when overlapping transactions occurred. Suppose initially s1/x=0 at server s1 and s2/y=0 at server s2. Client1 updates s1/x=10 and s2/y=10 at transaction v1, and at the same time client2 updates s1/x=20 and s2/y=20 at v2. Suppose the clients contact the servers in different order and the update messages arrive at the servers in reverse order, such that s1's pending queue is [x.v1=10, x.v2=20] and s2's pending queue is [y.v1=20, y.v2=10]. After client1 and client2 send the make-good command, s1's good queue is [x.v1=10, x.v2=20] and s2's good queue is [y.v2=20, y.v1=10]. When a third client3 tries to read the latest value of x or y, what is the latest value of its peer data? It looks like depending which data client3 starts with, it would get a different version of the peer data? Like starts with s1/x got s1/x.v2 == 20 and s2/y.v2 == 20, while starts with s2/y.v1 == 10 and s1/x.v1 == 10. Am I missing something or this is the semantic in determining the latest values of peer data?

pbailis 13y ago
(edit: post author, [not OP]) here. Thanks for the feedback! > I guess one of the key insights is that each data has a canonical server owner which enforces the consistency of the writes of the data at a single place. Well, this is the way I presented it, because it's easiest to understand. But, if you want a replicated system that provides HA, there are alternatives (http://www.bailis.org/blog/non-blocking-transactional-atomicity/#ha-note http://www.bailis.org/blog/non-blocking-transactional-atomic...). > When a third client3 tries to read the latest value of x or y, what is the latest value of its peer data? It looks like depending which data client3 starts with, it would get a different version of the peer data? ... Am I missing something or this is the semantic in determining the latest values of peer data? Good question! This ultimately comes down to how you want to handle concurrent writes. Many distributed databases use a "last writer wins" strategy when reconciling concurrent updates (that is, the correct behavior is specified to be that the database serves the highest-timestamped version of a given data item). Now, in your example, the clients both started their writes at the same (real-world) time and used this time as a basis for their timestamp, so the "last" write is undefined. We need a way to break the "last writer wins" tie. In practice, this can be something like the client ID appended to the last few bits of the timestamp or even a hash of the value written--as long as the tie-breaker is deterministic (that is, different replicas don't decide different "winners"), it doesn't really matter which is chosen. In practice, to avoid storing every value ever written, you'd want to provide some kind of "merge" function for multiple writes, and, in our implementation and in often practice, this is last writer wins (plus some deterministic tie-breaker for identically timestamped but distinct writes).

dscrd 13y ago

I think there's a typo under "good, pending and Invariants"? Isn't the caption supposed to be "not okay / (x=1 should be in pending or good)"

pbailis 13y ago
Post author here. Hmm. The invariant we're trying to maintain is that any write in 'good' should have its transactional "siblings" in either 'good' or 'pending' on their respective servers. So if we are trying to write x=1 and y=1, then, if x=1 is in the x server's 'good', then y=1 should be in the y server's 'good' or 'pending'. But in the example under "not okay", y is not present in either.
- dscrd 13y ago
  Ok, then I just did not understand that part yet.

yxhuvud 13y ago

Getting only 4.8% percent of throughput for reads doesn't seem stellar. Why is this so bad?

pbailis 13y ago
Ha--good catch! I don't know if anyone else noticed that. I meant to say "within 33-4.8% of the peak throughput..." That is, we're 95.2% of the peak, not 4.8% of the peak. Fixed, thanks!

espeed 13y ago

This model is somewhat similar to how Datomic models transactions -- Datomic uses immutable datoms and includes a transaction ID with each datom: "A datom consists of an entity, attribute, value and transaction (time)" (http://www.datomic.com/rationale.html http://www.datomic.com/rationale.html). However, Datomic's design is based on a single, centralized transactor that pushes out all the new transaction information to an index that is distributed to all its clients (peers), whereas NBTA would enable distributed transactors. See "The Datomic Data Model" http://www.datomic.com/rationale.html#DataModel http://www.datomic.com/rationale.html#DataModel From Datomic's FAQ http://www.datomic.com/faq.html http://www.datomic.com/faq.html ... How does Datomic provide ACID guarantees? The transactor serializes all transactions, and each transaction runs against a stable view of the database, and succeed or fail in their entirety. Transactions are not acknowledged until after they are logged in storage. Atomic read/modify/write operations can be performed by database functions, which run on the transactor within the transaction. Note that Datomic can provide ACID guarantees without utilizing read-transactions, nor read locks, due to the presentation to the query engine(s) of the database as an immutable value. ---//--- Rich Hickey on Datomic's design... Intro to Datomic (12 min) http://www.youtube.com/watch?v=RKcqYZZ9RDY# http://www.youtube.com/watch?v=RKcqYZZ9RDY#! The Design of Datomic (60 min) http://www.infoq.com/presentations/The-Design-of-Datomic http://www.infoq.com/presentations/The-Design-of-Datomic The Datomic Architecture and Data Model (46 min) https://vimeo.com/45136212 https://vimeo.com/45136212

pbailis 13y ago
Very cool, thanks! In general, immutable data items makes NBTA much easier; there's only one element in either 'good' or 'pending' for each data item at a time. The benefit of a single centralized transactor/ID generator is that it provides a simple total ordering on updates. The downside is that updates must (often) complete sequentially to guarantee safety and that the centralized service can become a bottleneck/challenge during failures. That said, I'm a big fan of the Datomic vision and look forward to seeing what they're able to accomplish.

agilord 13y ago

In theory this idea looks OK. I'd be interested if someone implemented it on Riak or Cassandra? What are the conclusions?