I continue to want to dabble here-abouts, since at least the time of having seen "Anna: A Crazy Fast, Super-Scalable, Flexibly Consistent KVS"[1], which is now one[2] of the two major components of the Hydro project, alongside Cloudburst[3], a Python "stateful serverless platform". There's also some other pieces like a local Anna-cache[4] & a coordinated retry library, & kubernetes-centric cluster-manager.
I have no idea how related the Hydro-Project work is to this recent publication on Calm, but I hope to find out!
I think this is highly misleading. They state that they achieved high speed-ups from not doing any coordination because they encode each state as a "private field" in an actor. Actors then communicate via queues (assume ZeroMQ from their reference), which require coordination to scale at some point - so do the actors that contain the actual data.
I do see some market for (eventual/in)-consistent databases though, so maybe that's something we are going to see much more often in the future.
I've read this paper a couple of times without having learned everything. The modeling language Bloom looks to have died on the vine. It's a datalog descendant. Bloom is supposed to help tell one if their distributed system is calm compliant
I made my own distributed database on top of my own HTTP app. server and I learned 2 things:
1) CRDTs and Paxos like protocols don't make any sense. Just send all data everywhere with both server and client async. transport and sync. "done" latch and your reads will load balance, you'll get 100% read uptime and "free" backup. That's about it.
2) Java can do joint (on shared memory) parallel stuff really well because of the VM GC and assembly cannot reclaim shared concurrent memory easily without those!
Here is my implementation proved with a MMO with 250.000 players: http://root.rupy.se
You're essentially describing a two-phase commit ie. commit the data to each node, and then to the whole cluster (or discard the transaction if not on every node).
This is fine for some distributed tasks, but you have to wait for the whole cluster before you can use the committed data - not fast compared to single-phase commits with eventual consistency.
This is what CRDTs, Paxos and CALM are trying to solve, a single phase commit where each node can then work out the order of transactions themselves, in order to reach an identical state to the next node.
In your MMO case, it looks like an ideal problem for 2-phase commits to solve :)
There's also a related problem, where the volume of data you're sending is such that you can't replicate it across the whole cluster, only some of it. You then have to a) sort out the redundancy of the data, and b) the consistency of the data bearing in mind that a node may have data reliant on data from another node.
Many of these issues are somewhat related to GC, distributed hash (merkle) trees, CRDTs and actors operating on data (rather than the other way round). It's a rich problem area with many coordination applications for the good solutions!
You don't need speed when your transport is completely non-blocking async. end-to-end because your CPUs are not IO-waiting on anything.
This scales way better and the time it takes to distribute over my global cluster is very little above the latency between the furthest nodes (~200ms).
What user case do they have for a distributed system that is faster than 200ms for global persistent data?
200ms for a registration or purchase is not huge?
You can obviously use the data before the transaction is complete anyway?!
I think Java is going to have a very strong revival on the server side once the Google lawsuit is over.
The whole point of CAP is not what happens when everything is up and talking to each other; the point is what happens when things start to fail (the 'P').
Let's go with two nodes (since you have no quorum, that's fine), A and B. A takes a write. Tells B about it.
Now what happens?
If B commits, acks to A, and -the network fails-, so that A never receives the ack (and thus says "didn't write"), you have a data split. B has committed data A has rejected as failed (and this of course becomes more and more likely the more distributed nodes that you have).
How do you resolve that? Maybe you don't; it might not matter. A future write (including a retry, since the writing node knows the distributed state is 'indeterminate') might update both, and you'll never notice. Even if a read is stale, it might not be noticed, or if it is just be shrugged away as a "can't reproduce". Of course, that assumes that B is reachable before you notice an issue; if it's a long term netsplit, A can never write over to B, and depending how you handle that you either can't take a write, or you allow B to grow very stale, and yet not -know- that it's stale (so you presumably are still allowing reads from it).
An eventually consistent scheme would accept and attempt to resolve that disparity; in fact, given that the data most likely to be noticed is user data, and users are likely only talking to one node at a time, you likely have an ideal case for eventual consistency.
In practice preparing for that kind of failure is a complete waste of time because you can never guarantee anything, the world is chaos and only "dead trees" (oil, coal and gas) make us believe that we are perfect.
In the 4 years of running this MMO we have had less than 0.3% refund because of network problems (including antivirus blocking of real-time messages, the MMO also uses HTTP). Personally I have never seen this database fail, it's much closer to perfect than any other distributed database if you count performance per watt and total cost of operation.
"In practice preparing for that kind of failure is a complete waste of time because you can never guarantee anything"
Oh, but you can. Hence the decades of research. And determining exactly what can be guaranteed, when, and what the tradeoffs are, is the whole point.
You're using refunds as a proxy for system correctness. That may make sense for your particular business case, but it doesn't make sense for many others. And it definitely isn't sufficient for a generic technical implementation that has to be able to be evaluated for an arbitrary business case.
Oops my bad, didn't look very closely at the repo. I think the gist of my text stands though, and as you've said it's not resilient to partition with multi-node updates, rollback or deletions (tombstones).
So I've commented more directly elsewhere, but I just want to point out that either A. You are correct, that distributed consensus is really simple and that you have stumbled upon an obvious truth that obviates decades of research by some of the brightest minds in this space, or B. You are incorrect, don't appreciate the complexity of the problem, that because you're blind to the the sharp edges believe they don't exist, and so look rather silly to anyone who -does- have knowledge in this space.
As I mentioned elsewhere, your use case likely mitigates the network failures. A given user is likely talking to only one node; you aren't getting writes to one, reads to another, let alone conflicting writes across nodes. Nor are you dealing with anything where inconsistencies create real problems, such as dealing with real world money (your users are more likely to go "Urgh, I thought I had enough X for that! Guess I have to go farm some more", not "WHERE THE HELL IS MY MONEY?! I DEPOSITED $1k LAST NIGHT, WHERE IS IT !?"). Nor have you solved your core problem better than an eventually consistent system would have (though possibly in a more energy efficient way, per your claim).
There's also a bit of irony asking "what do (I) think", and then objecting that my comments are made 'without arguments'. No, there are pretty clear arguments, to which your only response is "well, I have operated a successful MMO". I never said you didn't, nor that a DB that handles partitions incorrectly can't be useful (hell, I've written production systems reliant on Mnesia in Erlang, which in the event of a netsplit just...splits, and stays split, because it fit out use case). Just that your claim that "CRDTs and Paxos like protocols don't make any sense", and that just doing what you did is sufficient for any distributed system, is reflective of you, not of reality.
Erlang, Go and Rust are completely meaningless languages because they cannot share memory efficiently, ie. they cannot operate one-to-many, many-to-one or many-to-many efficiently.
So I value your arguments about "people have researched this" like I do with all people that don't get their hands dirty and prove things practically first, because obviously you have not.
I use real data from the real world, you should try it, you might invent something.
But yes my system relies on being connected to one node at the time like you said previously.
Barely going to touch those language claims; I've written production systems in two of those, and Java, your preferred stack. They all can share memory efficiently between units of concurrency; this just seems like another example of you claiming your own limited knowledge as being truth.
In terms of building real world systems - oh, but I have. Over a decade of building distributed systems, across 5 different companies. In domains that cared about correctness. You have a data point; a weak one. "This worked for me". Which I'm not arguing against (well, I'm nuancing it a bit; it worked sufficiently well to not cause business issue, which is a different metric than system correctness, but somewhat immaterial).
What I'm arguing against is your claim that "This will work for everyone" and "everything else is too complicated".
Which you've been unable to refute. In fact, you've been unable to even provide evidence for your own position (beyond the aforementioned "it worked for me"), let alone the counter claim.
My preferred stack is Java for server side, but C for the client. All other languages are completely flawed in too many ways that any sane person would choose to use them other than by mistake.
I have given you the source of a 2000 line distributed database that out-scales all other databases for 80% of their uses, and all you did was to down vote me.
I'm just saying "here is what I learned", you have thought me nothing.
Another way to think of a distributed system is as a threaded application on a multi-core CPU. Break down the interactions within the application by all the logical boundaries and you have a basic distributed system. Typically the input and output are via a single controlled mechanism and state is handled in a single place in memory during the series of events for a particular operation, so consistency is just assumed. But give all the series of events their own state, and each event a different input and output in random ordering, and you have distributed system problems. Threaded apps have similar problems to solve and have their own mechanisms to deal with these problems, so you can compare them and see how multiple mechanisms may be needed to compute safely.
You're right, it's the same synchronisation/consistency problem at a different scale, which I honestly find pretty mindblowing. However there is one major difference. When you add fault tolerance requirements (which you typically do in distributed databases and web applications, but don't always in multithreaded applications) then it changes the approaches and algorithms.
I know work is ongoing in the CRDT space (Kleppman's work gets posted here regularly). I'm assuming Hellerstein et al. are continuing work on distributed consistency, but "what and where" is not immediately obvious . Bloom/bud is being maintained, but that seems to be about it.
It seems like event-sourcing draws a tiny bit from this well, but it could be much more powerful with some help from CALM or similar.
Can anyone point to any other areas of active research and development in the area of reducing conflict in distributed systems?
22 comments
[ 2.9 ms ] story [ 69.0 ms ] threadI have no idea how related the Hydro-Project work is to this recent publication on Calm, but I hope to find out!
[1] https://rise.cs.berkeley.edu/blog/anna-kvs/
[2] https://github.com/hydro-project/anna
[3] https://github.com/hydro-project/cloudburst
[4] https://github.com/hydro-project/anna-cache
I do see some market for (eventual/in)-consistent databases though, so maybe that's something we are going to see much more often in the future.
1) CRDTs and Paxos like protocols don't make any sense. Just send all data everywhere with both server and client async. transport and sync. "done" latch and your reads will load balance, you'll get 100% read uptime and "free" backup. That's about it.
2) Java can do joint (on shared memory) parallel stuff really well because of the VM GC and assembly cannot reclaim shared concurrent memory easily without those!
Here is my implementation proved with a MMO with 250.000 players: http://root.rupy.se
Please comment if you down vote.
This is fine for some distributed tasks, but you have to wait for the whole cluster before you can use the committed data - not fast compared to single-phase commits with eventual consistency.
This is what CRDTs, Paxos and CALM are trying to solve, a single phase commit where each node can then work out the order of transactions themselves, in order to reach an identical state to the next node.
In your MMO case, it looks like an ideal problem for 2-phase commits to solve :)
There's also a related problem, where the volume of data you're sending is such that you can't replicate it across the whole cluster, only some of it. You then have to a) sort out the redundancy of the data, and b) the consistency of the data bearing in mind that a node may have data reliant on data from another node.
Many of these issues are somewhat related to GC, distributed hash (merkle) trees, CRDTs and actors operating on data (rather than the other way round). It's a rich problem area with many coordination applications for the good solutions!
This scales way better and the time it takes to distribute over my global cluster is very little above the latency between the furthest nodes (~200ms).
What user case do they have for a distributed system that is faster than 200ms for global persistent data?
200ms for a registration or purchase is not huge?
You can obviously use the data before the transaction is complete anyway?!
I think Java is going to have a very strong revival on the server side once the Google lawsuit is over.
I cannot explain the performance I'm getting yet but I'm working on that: https://github.com/tinspin/rupy/wiki (Under "Why Rupy is using Java")
The whole point of CAP is not what happens when everything is up and talking to each other; the point is what happens when things start to fail (the 'P').
Let's go with two nodes (since you have no quorum, that's fine), A and B. A takes a write. Tells B about it.
Now what happens?
If B commits, acks to A, and -the network fails-, so that A never receives the ack (and thus says "didn't write"), you have a data split. B has committed data A has rejected as failed (and this of course becomes more and more likely the more distributed nodes that you have).
How do you resolve that? Maybe you don't; it might not matter. A future write (including a retry, since the writing node knows the distributed state is 'indeterminate') might update both, and you'll never notice. Even if a read is stale, it might not be noticed, or if it is just be shrugged away as a "can't reproduce". Of course, that assumes that B is reachable before you notice an issue; if it's a long term netsplit, A can never write over to B, and depending how you handle that you either can't take a write, or you allow B to grow very stale, and yet not -know- that it's stale (so you presumably are still allowing reads from it).
An eventually consistent scheme would accept and attempt to resolve that disparity; in fact, given that the data most likely to be noticed is user data, and users are likely only talking to one node at a time, you likely have an ideal case for eventual consistency.
In the 4 years of running this MMO we have had less than 0.3% refund because of network problems (including antivirus blocking of real-time messages, the MMO also uses HTTP). Personally I have never seen this database fail, it's much closer to perfect than any other distributed database if you count performance per watt and total cost of operation.
Oh, but you can. Hence the decades of research. And determining exactly what can be guaranteed, when, and what the tradeoffs are, is the whole point.
You're using refunds as a proxy for system correctness. That may make sense for your particular business case, but it doesn't make sense for many others. And it definitely isn't sufficient for a generic technical implementation that has to be able to be evaluated for an arbitrary business case.
Please stop down voting without arguments, you are making the world a lesser place.
There's also a bit of irony asking "what do (I) think", and then objecting that my comments are made 'without arguments'. No, there are pretty clear arguments, to which your only response is "well, I have operated a successful MMO". I never said you didn't, nor that a DB that handles partitions incorrectly can't be useful (hell, I've written production systems reliant on Mnesia in Erlang, which in the event of a netsplit just...splits, and stays split, because it fit out use case). Just that your claim that "CRDTs and Paxos like protocols don't make any sense", and that just doing what you did is sufficient for any distributed system, is reflective of you, not of reality.
So I value your arguments about "people have researched this" like I do with all people that don't get their hands dirty and prove things practically first, because obviously you have not.
I use real data from the real world, you should try it, you might invent something.
But yes my system relies on being connected to one node at the time like you said previously.
In terms of building real world systems - oh, but I have. Over a decade of building distributed systems, across 5 different companies. In domains that cared about correctness. You have a data point; a weak one. "This worked for me". Which I'm not arguing against (well, I'm nuancing it a bit; it worked sufficiently well to not cause business issue, which is a different metric than system correctness, but somewhat immaterial).
What I'm arguing against is your claim that "This will work for everyone" and "everything else is too complicated".
Which you've been unable to refute. In fact, you've been unable to even provide evidence for your own position (beyond the aforementioned "it worked for me"), let alone the counter claim.
I have given you the source of a 2000 line distributed database that out-scales all other databases for 80% of their uses, and all you did was to down vote me.
I'm just saying "here is what I learned", you have thought me nothing.
It seems like event-sourcing draws a tiny bit from this well, but it could be much more powerful with some help from CALM or similar.
Can anyone point to any other areas of active research and development in the area of reducing conflict in distributed systems?