-
Notifications
You must be signed in to change notification settings - Fork 3
System Design and Architecture
Some random thoughts and explanations for design decisions regarding the overall architecture.
Chinese Whispers uses UDP unicast for the underlying communication protocol. This choice has a profound effect on the overall architecture because of the design limitations imposed by using datagrams as one's network transport. I chose UDP because I'm a strong believer in constraints. One of the biggest mistakes one can make with command and control for distributed systems is pushing too much state. State costs an awful, awful lot to maintain across a distributed system. Consequently, any distributed command and control should be ruthless in minimizing the amount of state maintained.
One of the surest ways to accomplish this design constraint is to simply not make it possible to pack in excessive state from the ground up. If you simply cannot pack in one meg of distributed state, then no one is going to be able to do such. That which is not forbidden is allowed and thus I design systems to forbid certain actions. Thus, UDP unicast makes an excellent constraint.
And UDP has a host of other extremely nice properties. First, it's fast. There is, by definition, no connection set up. Thus, one can communicate with new members without a lot of expensive session establishment time. The system, therefore, is extremely reactive and doesn't have the characteristic sluggishness you find in TCP based systems. Since we have to operate within the MTU, all we have to work with is 1500 bytes and sending 1500 bytes is nothing for modern network. This means the system is extremely bandwidth efficient.
Another advantage that UDP unicast has is that we need only one socket. We can talk to any number of parters using only one socket. From a resource consumption point of view, this is a huge advantage. With TCP, you only have a connection between two endpoints. Consequently, what you end up with is N^2 connections. This number grows pretty huge as the number of members increases.
And thus UDP unicast based systems are far less complicated. The N^2 problem with TCP inevitably results in complicated connection caching schemes to try to minimize the vast number of connections required even in moderately sized system. These schemes are complicated, brittle and add even more latency to an already complicated system. More complicated means more likely to fail. More complicated means harder and harder to understand, which means more conceptual bugs at the architectural level. Not good.
With UDP, the communications are butt simple. The design forces a spartan attitude because you simply don't have much to deal with in the first place. Unless you're truly insane, it's hard to make anything more complicated either. Thus, the resulting design is far simpler, far easier to test, far easier to understand, and results in a higher performance, high scaling system than any TCP counterpart.
Which is a very good thing.
Note that the gossip protocol, itself, is extremely robust to missing packets. Given that the underlying strategy of gossip is random communication, this means that Chinese Whispers is extremely robust in the face of dropped UDP packets. The very nature of epidemic anti-entropy protocols ensures that there multiple routes to spread the information within the system and if one or two or more routes fail, other routes won't.
Which makes the system kind of un-killable.
Chinese Whispers is a hybrid system in that it combines both a gossip protocol and a ring formed of its members to quickly, cheaply and reliably replicate state changes throughout its member nodes.
The gossip protocol is based on the paper Efficient Reconciliation and Flow Control for Anti-Entropy Protocols. This protocol replicates state and forms both a member discovery and failure detection service. Periodically, the protocol chooses a random member from the system view and initiates a round of gossip with it. A round of gossip is push/pull and involves 3 messages.
For example, if node A wants to initiate a round of gossip with node B it starts off by sending node B a gossip message containing a digest of the view number state of the local view of the replicated state. Node B on receipt of this message sends node A a reply containing a list of digests representing the updated state required, based on the received digests. In addition, the node also sends along a list of updated state that is more recent, based on the initial list of digests. On receipt of this message node A sends node B the requested state that completes a round of gossip.
When messages are received, the protocol updates the endpoint's failure detector with the liveness information. If the endpoint's failure detector predicts that the endpoint has failed, the endpoint is marked dead and its replicated state is abandoned.
Chinese Whispers extends the gossip protocol by dynamically creating and maintaining a ring of the members that allows the system to quickly broadcast state changes throughout the system. However, due to the nature of the system, incomplete views of the system and differing views by members prevent this from being a completely foolproof broadcast system. Instead, the system leverages the ring structure as a supplement to random gossip of the system. Combined, the ring provides a high speed mechanism to spread new state throughout the system and the random gossip protocol to bootstrap and maintain the ring as well as fill in the inevitable gaps in a loosely coupled distributed system.
One of the main reasons why the system works as well as it does is the the way it uses time. As mentioned elsewhere, Chinese Whispers requires NTP to ensure that clock skew between nodes is minimal. We make use of time in the state replication logic, where we order state by system timestamp. However, all this really requires is that the owning node of the state doesn't have its clock whacked out - i.e. it doesn't need to be synchronized with any other node as long as its own clock is monotonic and any state changes happen at greater than the clock frequency (i.e. >= 1 millisecond).
However, the failure detectors in the system all have to deal with time as they all fundamentally work by timeout. Because gossip is random and quite erratic, clean, consistent heartbeats are rather impossible.
If there are no state changes on a node, then the liveness of the node is not propagated throughout the membership. Consequently, nodes can be determined to be dead even though they are quite alive. There's a number of ways to solve this, but a simple mechanism is to create a special internal heartbeat state and update that every N gossip cycles. Eventually, we're going to send additional meta information such as suspicion matrices so we can then commandeer the heartbeat state for this.
Failure detection in gossip systems is tricky. Unless the gossip is not randomized, the communication between members is purposefully randomized. The upshot is that if you're relying on intercommunication between endpoints to indicate the liveness of endpoints, you're going to be one frustrated puppy.
In Chinese Whispers, I take a different approach. The liveness of an endpoint is represented by the changing state that is replicated throughout the system. This is why all the replicated state has a timestamp instead of a checksum, which would serve the same purpose of indicating state change. When new state from node N shows up on node X, the failure detector uses the timestamp of the state as a proxy for the heartbeat.
This results in a kind of virtual heartbeat that propagates around the membership by virtue of very epidemics of the state replication gossip. The upshot is that endpoints that may not have heard from node M will learn of its replication of live state to some other endpoint that eventually replicated itself to node M. This allows Chinese Whispers to perform significantly better from a failure detection perspective than most any other heartbeat strategy.
But failure detectors are famously finicky, so even using replicated state timestamps as our heartbeat interval measurement isn't enough to calm down their jumpy souls. Thus, Chinese Whispers includes one more parameter to pass to the failure detection: the delay in receiving the heartbeat state. You can see this parameter used in the AdaptiveFailureDetector class (BTW, this is the failure detector I recommend you use. Really beats anything else all to shit, afaik).
The failure detector keeps a running average of all the delays seen by the detector and when asked to judge the failure probability of the endpoint, it uses this running average delay to adjust the sample time to compensate for the average time it takes for this endpoint's state to replicate throughout the membership to the node where the failure detection is being made.
Chinese Whispers also uses the standard technique of waiting for some time period to ensure that the failed endpoint still has failed. Again, due to the spiky, random nature of gossip communication and replication, and even using the state replication and weighting the detector with the average state propagation delay, we can still see false positives. Thus, Chinese Whispers uses a clean up cycle that indicates how many gossip cycles the endpoint has to fail before it is finally declared a dead parrot.
Finally, Chinese Whispers borrows a technique that I can't remember where I learned from (will update with a reference when I find it). The issue with accrual failure detectors (the kind that Chinese Whispers uses) is that they require a history of samples in order for their predictions to be accurate. It's the old GIGO problem. Again, randomized gossip is spiky and highly varying, particularly during times of high membership churn (such as at the establishment of the cluster).
Thus, what Chinese Whispers allows you to do is set up the failure detector with a synthetic history that will eventually be aged out as real samples fill up the failure detector's sliding window. The result is that - when combined with the other tricks mentioned above - the failure detection performance of Chinese Whispers is superb. Adjusting the failure detection parameters is rather intuitive and not the exercise in frustration that is standard fare for other gossip system's failure detection.
The thresholds you set are inevitably multiples of the gossip interval and easy to understand. You set your pain threshold, a little bit of slop and some multiple of the gossip interval to prime the detector and it just works. You don't have to balance yourself on a knife edge between life and death (so to speak), and the parameters invariably work across a wide range of real world conditions.