Yup, this is the key problem with these consensus algorithms. They don't handle shards very well. Most systems these days either punt and don't do updates via the consensus algorithm (Apache Kafka), or they use two-phase commit (FoundationDB, many others).
This is important work, but I don't think they're really going to succeed is making a truly large-scale system. The problem is the heartbeat; you can't have every node talking to every other node every few seconds.
What we really need is someone smart to come up with a consensus algorithm that doesn't need a heartbeat. Until then, it's two-phase commit if you want a reliable, large-scale (if not performant) system.
Right now, if you want a system that runs fast you use a consensus algorithm for shard metadata, but you do writes directly to the nodes without getting a consensus first. You run the risk of losing acknowledged writes, but it's the best you can do if you need speed.
(author of the original post here) Raft can be pretty relaxed about heartbeats. It doesn't, strictly speaking, require every node to talk to everyone else. All you really need is some health signal about the other nodes so you A) don't call an election while the leader is still alive and B) call an election promptly when a leader disappears. As a first step, we've reduced our heartbeat traffic from one per group to one per node, and we can reduce it further (e.g. by having each node send heartbeats only to a subset of its peers and sharing the results to other nodes).
Also, we are using both Raft and two-phase commit: Raft manages the consistency of individual ranges, but transactions that span multiple ranges require 2PC.
> Does handling all range's consensus traffic in bulk
would be the most effective way to handle this?
Instead could all the ranges from a node be represented
as a single state machine to be handled by plain etcd
raft implementation, like the approach taken in spanner paper?
Do you have any comment on this? I've been trying to wrap my head around the implications. What's the downside?
I replied on the blog post. The downside is basically that if you have one giant consensus group, then to add a new replica you have to replicate that entire consensus group. This hurts load balancing and recovery times, since you can't scatter a dead node's ranges across all the other nodes in the cluster.
> The problem is the heartbeat; you can't have every node talking to every other node every few seconds.
The node does not have to talk with EVERY other nodes.
> What we really need is someone smart to come up with a consensus algorithm that doesn't need a heartbeat.
Well. Most consensus does not require heartbeat. Heartbeat is just the simplest way to detect a potential failure.
> if you want a system that runs fast
Consensus might affect the commit latency. It wont affect throughput that much though. But if you do not care, you can simply write to a consensus system without waiting for the commit. You can still get the serialization and replication.
Please clarify what you mean by "these consensus algorithms". Raft is merely an instance of a specific subset of C.A.s (published) to date.
Raft's design goals were primarily to be "understandable". A noble pedagogical goal in the abstract but misguided for use in production given the prevailing mindset [1].
> What we really need is someone smart ...
[1] Under 20 years of age only or will you accept advice from 50+ set? /G
You are thinking about consistent RSM. Paxos and Raft are not the similar thing and not even at the same level.
Epaxos, along with other master-less RSM, does not require heartbeat. It is not about smart people, it is about how you want to implement your system based on the requirement.
I'm not sure how you could call it a consensus algorithm without putting updates through the state machine, since that's essentially all Raft does.
"The problem is the heartbeat; you can't have every node talking to every other node every few seconds"
So nodes are just to assume that other nodes are alive/reachable/functioning? How would you propose detecting node health without some form of heartbeat?
"... but you do writes directly to the nodes without getting a consensus first. You run the risk of losing acknowledged writes..."
This is the equivalent of the default MongoDB write strategy. You probably don't want it. In the majority of use cases you really, really want writes to be committed to the replicated state machine on a quorum of nodes before they are acknowledged back to the client.
> So nodes are just to assume that other nodes are alive/reachable/functioning? How would you propose detecting node health without some form of heartbeat?
generally there is probably enough traffic between nodes to just run failure detection off of txn failures / normal comms (rpc,etc) without running a separate failure detection mechanism, but this assumes every node is already talking to every node, which is not safe, but let's explore solutions to that. detecting failures from normal communication will likely be even faster than a heartbeat, provided traffic frequency is higher than the heartbeat (use case contingent). to get around the issue where all nodes aren't already in communication, could broadcast any local failures each node might observe and you save the traffic of the heartbeat in the normal case, a gossip mechanism could work to reduce the N^2'ness of that communication -- it seems likely that multiple nodes will detect a given failure at around the same time, we don't want each of them broadcasting that, potentially leading to more failures from increased net traffic. since the nodes who were in communication with the failed one have detected it, we really just need to update the nodes who weren't so that they can find out about it in a closed interval (same guarantee as a heartbeat, potentially looser bounds).
maybe this is being too optimistic, but at least it's fun to think about :)
Heartbeating is only used for failure detection. Because raft is a consensus mechanism, failure detection is allowed to return false positives. In practice false positives cause a blip in throughout as a new leader gets elected, but since they are rare it isn't a big problem.
Because of the weak guarantees needed by the failure detector, you can use efficient mechanisms like gossip to reduce the bandwidth required. You pay some detection latency for that in return. It sounds like cockroachdb doesn't do that right now, but they could. It isn't some fundamental problem with consensus systems (it is a completely separate problem).
12 comments
[ 3.0 ms ] story [ 34.0 ms ] threadThis is important work, but I don't think they're really going to succeed is making a truly large-scale system. The problem is the heartbeat; you can't have every node talking to every other node every few seconds.
What we really need is someone smart to come up with a consensus algorithm that doesn't need a heartbeat. Until then, it's two-phase commit if you want a reliable, large-scale (if not performant) system.
Right now, if you want a system that runs fast you use a consensus algorithm for shard metadata, but you do writes directly to the nodes without getting a consensus first. You run the risk of losing acknowledged writes, but it's the best you can do if you need speed.
Also, we are using both Raft and two-phase commit: Raft manages the consistency of individual ranges, but transactions that span multiple ranges require 2PC.
> Does handling all range's consensus traffic in bulk would be the most effective way to handle this? Instead could all the ranges from a node be represented as a single state machine to be handled by plain etcd raft implementation, like the approach taken in spanner paper?
Do you have any comment on this? I've been trying to wrap my head around the implications. What's the downside?
The node does not have to talk with EVERY other nodes.
> What we really need is someone smart to come up with a consensus algorithm that doesn't need a heartbeat.
Well. Most consensus does not require heartbeat. Heartbeat is just the simplest way to detect a potential failure.
> if you want a system that runs fast
Consensus might affect the commit latency. It wont affect throughput that much though. But if you do not care, you can simply write to a consensus system without waiting for the commit. You can still get the serialization and replication.
Raft's design goals were primarily to be "understandable". A noble pedagogical goal in the abstract but misguided for use in production given the prevailing mindset [1].
> What we really need is someone smart ...
[1] Under 20 years of age only or will you accept advice from 50+ set? /G
(I'm 53. Would be happy to hear your old-man theories :)
Epaxos, along with other master-less RSM, does not require heartbeat. It is not about smart people, it is about how you want to implement your system based on the requirement.
"The problem is the heartbeat; you can't have every node talking to every other node every few seconds"
So nodes are just to assume that other nodes are alive/reachable/functioning? How would you propose detecting node health without some form of heartbeat?
"... but you do writes directly to the nodes without getting a consensus first. You run the risk of losing acknowledged writes..."
This is the equivalent of the default MongoDB write strategy. You probably don't want it. In the majority of use cases you really, really want writes to be committed to the replicated state machine on a quorum of nodes before they are acknowledged back to the client.
generally there is probably enough traffic between nodes to just run failure detection off of txn failures / normal comms (rpc,etc) without running a separate failure detection mechanism, but this assumes every node is already talking to every node, which is not safe, but let's explore solutions to that. detecting failures from normal communication will likely be even faster than a heartbeat, provided traffic frequency is higher than the heartbeat (use case contingent). to get around the issue where all nodes aren't already in communication, could broadcast any local failures each node might observe and you save the traffic of the heartbeat in the normal case, a gossip mechanism could work to reduce the N^2'ness of that communication -- it seems likely that multiple nodes will detect a given failure at around the same time, we don't want each of them broadcasting that, potentially leading to more failures from increased net traffic. since the nodes who were in communication with the failed one have detected it, we really just need to update the nodes who weren't so that they can find out about it in a closed interval (same guarantee as a heartbeat, potentially looser bounds).
maybe this is being too optimistic, but at least it's fun to think about :)
Because of the weak guarantees needed by the failure detector, you can use efficient mechanisms like gossip to reduce the bandwidth required. You pay some detection latency for that in return. It sounds like cockroachdb doesn't do that right now, but they could. It isn't some fundamental problem with consensus systems (it is a completely separate problem).