Time, Clocks and Ordering in a Distributed System

A distributed system consist of a collection of distinct processes which are spatially separated, and which communicate via messaging. Here considered the concept of events ordering in a distributed systems. First, discussed the partial ordering defined by "happened before" relation and give a distributed algorithm for extending it to a consistent total ordering of all events.

Partial Ordering

The relation "happened before" (->) on the set of events of a system is the smallest relation satisfying the following three conditions:

• If a and b are events in the same process, and a comes before b, then a -> b.
• If a is the sending of a message by one process and b is the receipt of the same message by another process, then a -> b.
• If a -> b and b -> c then a -> c.

Two distinct events a and b are said to be concurrent if a !-> b and b !-> a.

The "happens before" relation describes partial ordering of the events in the system.

Logical Clock can be represented by the following condition (it doesn't specifically means physical clock):

      Clock Condition. For any events a, b: if a -> b then C(a) < C(b). 

Note that converse condition doesn't hold so if C(a) < C(b) then we cannot expect that a -> b and two cases are possible either a -> b or a and b are concurrent events. Which means that by knowing just the clock values you cannot distinguish concurrent events.

From the definition of "happens before", Clock Condition is satisfied if the following two conditions hold:

• C1. If a and b are events in process Pi, and a comes before b, then Ci(a) < Ci(b).
• C2. If a is the sending of a message by process Pi and b is the receipt of that message by process Pj, then Ci(a) < Cj(b).

Implementation rules:

• IR1. Each process Pi increments Ci between any two successive events.
• IR2. (a) If event a is the sending of a message m by process Pi, then the message m contains a timestamp Tm = Ci(a). (b) Upon receiving a message m, process Pj sets Cj greater than or equal to its present value and greater than Tm.

Java implementation of Clock Condition as an illustration to this post can be found at Logical Clocks repository. The base classes are LogicalTimestamp which represent specific moment of time and LogicalClock class which provides thread-safe methods to store and update time according to implementation rules and conditions described above during events at each process of distributed system. 

Total Ordering

We can use a system of clocks that satisfying Clock Condition to define a total ordering (=>).
If a is an event in process Pi and b is an event in process Pj, then a => b if and only if either:

• (i) Ci(a) < Cj(b) or 
• (ii) Ci(a) = Cj(b) and Pi < Pj. 

We can use use the concept of total ordering to solve simple synchronization problem. We wish to find the algorithm for granting access to shared resource to the process which satisfy following conditions:

• (I) A process which has been granted resource must release it before it can be granted to another process;
• (II) Different requests for the resource must be granted in the order in which they are made;
• (III) If every process which is granted the resource eventually release it, then every request is eventually granted.

Algorithm which satisfy conditions I-III and uses concept of total ordering can be described with the following 5 rules:

1. To request the resource, process Pi sends the request resource message Tm:Pi to every other process, and puts that message on its request queue, where Tm is the timestamp of the message. 
2. When process Pj receives the request resource message Tm:Pi, it places it on its request queue and sends a (timestamped) acknowledgment message to Pi. 
3. To release the resource, process Pi removes any Tm:Pi request resource message from its request queue and sends a (timestamped) Pi release resource message to every other process.
4. When process Pj receives a Pi release resource message, it removes any Tm:Pi request resource message from its request queue.
5. Process Pi is granted the resource when the following two conditions are satisfied: (i) There is a Tm:Pi request resource message in its request queue which is ordered before any other request in its queue by the total ordering relation (=>); (ii) Pi has received a message from every other process timestamped later than Tm.

It is a distributed algorithm. Each process independently follows these rules to perform an emergent function (in this case granting synchronized access to the shared resource) , and there is no central coordinating process or central storage.

Such approach and principles can be generalized and allow to implement any desired form of multiprocess synchronization in a distributed system. But it depends from operating of all processes and failure of one process will halt other processes. The problem of failure not considered here. 

Anomalous Behavior

There can be an anomalous behaviour if some events which satisfy condition a -> b happens outside the system then system can assign resource access in the wrong order since from the point of view of the system a !-> b which can contradict to expected behaviour by the users. This can be prevented by the use of properly synchronized physical clocks. It can be formally shown how closely the clocks can be synchronized.

References:

1. Time, Clocks and the Ordering of Events in a Distributed System (1978)
2. Logical Clocks github repository

Scalable and Efficient Distributed Failure Detectors

Failure detectors are a central component in fault-tolerant distributed systems running over unreliable asynchronous networks e.g., group membership protocols, supercomputers, computer clusters etc. The ability of the failure detector to detect process failures completely and efficiently, in the presence of unreliable messaging as well as arbitrary process crashes and recoveries, can have a major impact on the performance of these systems

Properties of failure detector:

• {Strong/Weak} Completeness: is the guarantee that failure of group member will be eventually detected by {all/some} non-faulty members.
• Strong Accuracy: no non-faulty group member is declared as failed by any other non-faulty member.

It is proved that achieving both strong completeness and strong accuracy is impossible on fault-prone networks. So if required strong completeness then need to accept weak accuracy (some possibility of false positive failure detection) while trying to reduce it to minimum.

Requirements to Failure Detector:

1. Completeness: satisfy eventual Strong Completeness.
2. Efficiency:

• Speed: quick (within some given time T) detection of member failure by some (not all) non-faulty member
• Accuracy: low probability (below given PM(T) which is much below of probability of message loss Pml) of failure detection mistakes.

 3. Scalability: equal expected worst-case network load per member

Theorem. Any distributed failure detector algorithm for a group of size n (>> 1) that deterministically satisfies the Completeness, Speed and Accuracy requirements above, for a given values of T and PM(T) (<< Pml), imposes a minimal worst-case network load (messages per time unit, as defined above) of:

Furthermore, there is a failure detector that achieves this minimal worst-case bound while satisfying the Completeness, Speed, Accuracy requirements. L* is thus the optimal worst-case network load required to satisfy the Completeness, Speed, Accuracy requirements.

Heartbeat-based approaches provides completeness, but have shortcomings:

• Centralized - creates hot-spots and prevent them from scaling
• Distributed - inherently not very efficient and scalable.

Randomized Distributed Failure Detector Algorithm:

It takes as assumption that list of members are same and already known on the nodes. Member recovers from failure in distinguishable new incarnation. Each message also contains current incarnation number of the sender.

At each member Mi:

Integer pr; /* local period number */

Every T' time units at Mi:

0. pr := pr + 1
1. Select random member Mj from view.

     Send a ping (Mi, Mj, pr) message to Mj
     Wait for the worst case message round trip time for an ack(Mi, Mj, pr) message.
2. If not received ack yet:

     Select k members randomly from view.

     Send each of them a ping-req(Mi,Mj, pr) message

     Wait for an ack (Mi, Mj, pr) until the end of period pr.
3. If not received ack(Mi, Mj, pr) message yet then declare Mj as failed.

Anytime at Mi:

4. On receipt of a ping-req(Mm,Mj, pr) (Mj != Mi)

     Send a ping(Mi, Mj, Mm, pr) message to Mj

     On receipt of an ack(Mi, Mj, Mm, pr) message from Mj

     Send an ack(Mm, Mj, pr) message to Mm

Anytime at Mi:

5. On receipt of a ping(Mm, Mi, Ml, pr) message from member Mm

     Reply with an ack(Mm, Mi, Ml, pr) message to Mm

Anytime at Mi:

6. On receipt of a ping(Mm, Mi, pr) message from member Mm

    Reply with an ack(Mm, Mi, pr) message to Mm

Fig. 1 Example of failure detection protocol period at Mi. This shows all the possible messages that a protocol period may initiate.

Parameters T' and k of the algorithm can be formally calculated based on the given required quality of service parameters: average speed of failure detection T and accuracy PM(T). 

This algorithm has uniform expected network load at all members. The worst-case network load occurs when, every T' time units, each member initiates steps (1-6) in the algorithm. Steps (1, 6) involve at most 2 messages, while steps (2-5) involve at most 4 messages per ping-req target member. Therefore, the worst-case network load imposed by this protocol (in messages/time unit) is:

L = n * [2 + 4 * k] * 1/T'

Which means linear overall network load O(n) produced by running algorithm on all nodes and constant network load at one particular member independent from the group size n.

References:

1. On Scalable and Efficient Distributed Failure Detectors (2001)

Quality of Service of Failure Detectors

Basic Quality of Service (QoS) properties of Failure Detector are:

• How fast it detects failure
• How well it avoids false detection

Assuming that process crash is permanent or in other words recovered processes will be the new identities.

Primary QoS metrics of Failure Detector:

• Detection time (T_d). How long it takes to detect failure.
• Mistake recurrence time (T_mr). Time between two consecutive mistakes.
• Mistake duration (T_m). Time it takes for failure detector to correct mistake.

Derived metrics (can be computed from primary metrics):

• Average mistake rate.
• Query accuracy probability. Probability that failure detector's output is correct at random moment of time.
• Good period duration.
• Forward good period duration.

The defined metrics do not depend on implementation-specific features of failure detection algorithm and can be used to compare failure detectors.

QoS requirements to the Failure Detector can be expressed via primary metrics:

• Upper bound on the detection time (T_ud)
• Lower bound on the average mistake recurrence time (T_lmr)
• Upper bound on the average mistake duration (T_um)

Together with probabilistic parameters of the network:

• Message loss probability [P_loss]
• Average message delay [E(D)]
• Variance of message delay [V(D)]. Exponential distribution of message delays is very common.

Also at the paper [1] was proposed a modified heartbeating algorithm for failure detection with input parameters:

• n -  delay between consecutive heartbeats
• ro - time shift of the heartbeat after which process declared as failed

The goal was to compute based on (1) QoS requirements (T_ud, T_lmr and T_um) and (2) the probabilistic parameters of network (P_loss, E(D) and V(D)), the optimal failure detector parameters n and ro for the proposed algorithm.

It is possible to estimate links quality parameters analyzing heartbeat messages. So we can adjust failure detector parameters dynamically using link quality estimator:

References:

1. On the Quality of Service of Failure Detectors (2002)

Fallacies of Distributed Computing

Above described assumptions which architects and designers of distributed systems are likely to make and which prove to be wrong in the long run:

• The network is reliable. The network is unreliable and we need to address possible failure scenarios.
• Latency is zero. Try to make few as possible network calls since latency is not zero and huge comparing to in-memory calls.
• Bandwidth is infinite. Try to simulate real production environment.
• The network is secure. You need to build security into your solution from Day 1. You need to be aware about security concerns and its implications even so the architect of the system should not be a security expert.
• Topology doesn't change. In real environment topology can change. And you need to be aware regards it (e.g. use “Next Hop” routing or specify address by DNS name). 
• There is one administrator. Administrators can constraint your options and you need to help them to manage your application.
• Transport cost is zero. There are costs associated with both computational resources and money spent on network maintenance.
• The network is homogeneous. Interoperability will be needed sooner or later. Do not rely on proprietary protocols but use standard technologies that are widely accepted.

References:

1. Fallacies of Distributed Computing Explained

Distributed vs. Non-Distributed Computing

Distributed and non-distributed computing has conceptual differences:

• Latency
• Memory access
• Partial failure
• Concurrency

Major problems in distributed computing correspond to this differences:

• Ensuring adequate performance
• Dealing with differences in memory models between local and distributed entities
• Dealing with partial failures and lack of a central resource manager
• Dealing with problems of concurrency

Distributed application interface should reflect its distributed nature. Merging this two models leads to one of the following problems:

• Making local computing looks like distributed makes local computing unnecessary difficult.
• Making distributed computing looks like local leads to the unreliable system.

References:

1. A Note on Distributed Computing (1994)