CIS 307: Logical (and physical) Clocks

Read Tanenbaum-vanSteen: 241-246, 252-261

We would like to order the events occurring in a distributed system in such a way as to reflect their possible causal connections. Certainly if an event A happens before an even B, then A cannot have been caused by B. In this situation we cannot say that A is the direct cause of B, but we cannot exclude that A might have influenced B. We want to characterize this "happened-before" relation on events. We will consider only two kinds of events, the sending of a message and the receiving of a message.

  1. If events e1 and e2 occur in the same system and e1 occurs before e2 (there is no problem to determine this in a single system) then e1 happened-before e2, written e1 -> e2
  2. If event e1 is the sending of a message and e2 is the receiving of that message, then e1 happened-before e2
  3. If e1 -> e2 and e2 -> e3 then e1 -> e3

The -> relation is a partial order. Given events e1 and e2 it is not true that either they are the same event or one happened-before the other. These events may be unrelated. Events that are not ordered by happened-before are said to be concurrent.

This characterization of happened-before is unsatisfactory since, given two events, it is not immediate (think of the time it takes to evaluate a transitive relation) to determine if one event happened before the other or if they are concurrent. We would like a clock C that applied to an event returns a number so that the following holds:

(1) If e1 -> e2 then C(e1) < C(e2)

We can define one such C, a logical clock (due to Lamport), as follows:

  1. On each computer i we start with a clock Li set at 0
  2. If e is the sending of a message, then increment the local clock Li, set C(e) to Li and timestamp the message with Li
  3. If e is the receiving at node i of a message with timestamp t, then set the local clock and C(e) to max{t, Li}+1
[We could have started the local clocks with their values set to the same value greater or equal to 0 and using as increments, instead of 1, any value greater than 0. The resulting time still would satisfy (1)] Here is an example showing events a,b,c,d,e,f,g,h,i,j and the corresponding local clocks:

Since the method we have chosen allows different events (on different machines) to have the same logical clock value, we extend the clock value with a number representing the machine where the event occurred [i, C(e)] and order these pairs according to the rule:

[i,C(e1)] < [j,C(e2)] iff C(e1) < C(e2) or [C(e1)=C(e2) and i<j]

Here is an example showing the same events but now tagged with numbers that totally order them:

Now that events are totally ordered, we could, for instance, maintain separate logs at each system, and then later combine the logs in a unique way. Also, we can use these timestamps to totally ordering multicast messages (i.e. all systems receive the multicasts in the same order) as we will see later.

Vector Clocks

A problem of the logical clocks considered above is that they do not represent faithfully the happened-before relation. Namely, while it is true that (1) holds, it is not true that

(2) If C(e1) < C(e2) then e1 -> e2

It would be nice to have logical clocks that would allow us to answer precisely questions of concurrency and dependency. We can do it by using clock values that are vectors of numbers, one for each computer system and then comparing vector values v and u with the rule

u < v iff ((for all components i, u[i]is less or equal to v[i]) and (there is a j such u[j]<v[j]))

Here are the rules on vector clocks in a system with n computers:
  1. Each computer starts with a local clock set at [0,0,..,0]
  2. When on computer i there is a sending event, increment the ith component of the clock by 1 leaving other components unchanged, then tag both the event and the message with this value
  3. When on computer i there is a receiving event, form a new local clock value taking the componentwise maximum of the local clock and the time stamp on the arriving message. Increment the ith component by 1. Finally tag the event with this value.
With these rules, in the vector clock v of an event e, V(e), at machine i, v[i] denotes the number of events that occurred at machine i up to e included, and for j, j!=i, v[j] denotes the number of events that occurred at machine j that are also known at machine i.

Here is our old example using vector clock values.

With vector clocks we have the property

e -> f if and only if V(e) < V(f)

Of course nothing is perfect. Though vector clocks allow to represent faithfully the happened before relation between events, they result in big time stamps and requires that we know in advance the number and identities of the sites involved.
Vector clocks can be totally ordered by ordering events with concurrent vector clocks on the basis of the machine where they occur.

All logical clocks, whether scalar or vector, suffer of a basic problem: though they represent that an even occurred (or not) before another, they do not represent before by how much, that is, there is no measure of elapsed time. In particular we cannot use logical clocks for timeouts, we need instead a wallclock (i.e. an old fashioned clock, possibly synchronized).
Another problem of logical clocks, even vector clocks, is that they cannot account for information exchanged outside of the computer systems, say because user A at system 1 receives output from the system and says that information to user B at system 2. And now user B enters that information at system 2.

Messages with FIFO Property and the Causal Ordering of messages

If we send two datagrams from host n1 to host n2 using the IP or UDP protocol we are not guaranteed that they will be received in the order they were sent. If instead we send data using the TCP protocol we are guaranteed that it will be received and it will be received in the order it was sent. Let's use the following notation, given a message m, SEND(m) is the event of sending the message m, and RECEIVE(m) is the event of receiving the message m. We say that a messaging system satisfies the FIFO property if: when m1 and m2 are messages sent from the same source S to the same destination D and we have SEND(m1) -> SEND(m2) then RECEIVE(m1) -> RECEIVE(m2). When this property is satisfied we have a natural ordering (total) for messages from S to D. The FIFO property is normally used in the context of multicast messages, where we want to be sure that messages sent from the same source are delivered everywhere in the same order.

We generalized FIFO ordering as follows:

Causal Ordering of Messages: m1 -> m2 iff (SEND(m1) -> SEND(m2)) implies !(RECEIVE(m2) -> RECEIVE(m1))


In going from the FIFO to the causal ordering we have dropped the requirement that messages come from the same source and that the receptions are related.
Note that the situation shown below in the diagram (A) is forbidden in a messaging system with the FIFO property, and the situation in diagram (B) is forbidden in a messaging system with the causal property, while the situation in diagram (C) is a causal ordering.


TCP assures the FIFO property on a single connection, but does not guaranty the causal property across multiple connections. Diagram (B) above is quite possible when using TCP.

Here is another time diagram, which we call the lost client diagram. 

Interpret the message m1 = (a,j) to mean "Give responsibility for client x to machine M2"
Interpret the message m2 = (b,c) to mean "Who is responsible for x?"
Interpret the message m3 = (d,e) to mean "Machine M2 is responsible for x"
Interpret the message m4 = (f,g) to mean "Connect me to x"
Interpret the message m5 = (h,i) to mean "I don't know x"
Thus at i machine M3 is in a state of confusion about the whereabouts of x.
This is clearly an undesirable situation.
Here are the vector clocks for the various events:
a = [1,0,0]	b = [0,0,1]	c = [2,0,1]	d = [3,0,1]	e = [3,0,2]
f = [3,0,3]	g = [3,1,3]	h = [3,2,3]	i = [3,2,4]	j = [3,3,3]
The lost client diagram does not satisfy the causal property: though a = SEND(m1) -> d = SEND(m3) we have e = RECEIVE(m3) -> j = RECEIVE(m1).

Making a multicast messaging system satisfy the causal property

Algorithms are available for converting a multicast messaging system without the causal property into one where the property is satisfied. One proceeds as follows.
The vector clock is computed to take into account only the sending of messages, not their reception (i.e. the local component of the vector clock is not incremented when a message is received). Thus a vector clock V on machine k will indicate at position k the number of messages sent by machine k, and at every other position j it will indicate the highest message received at k from j.
When node k with local vector clock v receives from node j a message m with vector timestamp u then the message will be delivered to the application when the following Delivery Rule holds:
  1. u[j] = v[j]+1, i.e. this message represents the first event at j occurring after the last j event known at k
  2. For all i != j, u[i] <= v[i], that is, node k knows now about events at other nodes at least as much as it was known at j when it sent the message m.
In diagram (B) above we see that the message at event f on machine 3 has timestamp [1,0,0] while the last event e on machine 3 had timestamp [2,1,0]. Thus we know that the message received at f was sent before the sending of the message received at e. This does not help us directly since by now the message at e has been accepted. But if we require that all messages be multicast to all participants, then each machine will be able to detect if a message is received out of order as shown in the following graph

now the message at e is received with timestamp [2,0,0] while the local clock is [0,0,0] thus we know that an event on machine 1 that preceded the sending of this message was not received on machine 3. Consequently we can postpone at machine 3 the delivery of the e message until after the f event has occurred.

Causal ordering of multicast messages guaranties that messages sent by the same system will be delivered everywhere in the same order they were sent. More in general if SEND(m1) -> SEND(m2) then m1 will be delivered everywhere before m2. But if SEND(m1) and SEND(m2) are concurrent, then there is no constraint on how these messages are delivered, so in one system m1 may arrive first, in another m2 may arrive first.

Totally Ordered Multicast

Here is an example from the textbook: A database with information on bank accounts is replicated in SF and NY. A teller in SF adds $100 to an account currently with $1000. A teller in NY simultaneously increases that account with a 1% interest. Depending on the order these updates are done the result is $1110 or $1111. [Either ordering will satisfy the Causal Ordering constraint.] Certainly we cannot live with a system where the updates can be done in different order at the replicated sites. The problem can be solved with totally ordered multicast: messages are delivered to all sites and delivery order is guaranteed to be the same at all sites.

Totally ordered multicasts can be implemented as follows: each site is connected to all others with 1-1 links. Each link has two properties: 1) it is reliable 2) messages are delivered in the order they were sent (FIFO property) . For instance TCP connections have these properties. [Alternatively the sites may use a form of multicast that satisfies those same two properties.] The sites use totally ordered logical clocks (say, vector + machine order to break ties). The sites keep queues.

  1. Each message, time stamped, is sent to all sites (multicast), including self.
  2. When a message is received, an acknowledgement is sent, time stamped with the same time as the message, to all sites, including self, and the message is placed in the local queue.
  3. A message that is in the queue and has been acknowledged by all the sites is said to be ready. Ready messages are extracted from the queue when not later that any non-ready message in the queue. They are then applied in the order of their clocks.
Does this algorithm work? Let's first observe that when a message m becomes ready at a site S, m has been received by all the sites since they have given an acknowledgement for m that was received at S. Second, if n is a message originated from the same site as m and sent before m, then everywhere n is received before m and the acknowledgements for n are received everywhere before the acknowledgements for m (this is so because of the properties assumed for links) thus n will be be ready before m and will be delivered to the application (be extracted from the queue) before m. So for each message originating site we need to worry only about one message from that site (the other messages from that site will be naturally delayed relative to the first one) If message n originates from a different site than m the argument is harder to make. It is possible that m and n are delivered in different order at different sites. But it is certain that before making the final decision to remove from queue either m or n, that both m and n and one acknowledgement of either m or n will be received at each site thus allowing the algorithm to compare clocks and possibly delay dequeueing in order to respect the clock total order. That is, in all cases, if n precedes m in the total order then n will become ready before m. Thus messages are delivered to the application in their clock order.
The following diagram illustrates the situation. Machine M2 sends message m and machine M3 sends message n. To simplify the diagram we show only the messages delivered to M1 and M4. The acknowledgements are indicated with a broken line and a primed name.

If m and n are delivered in different order at M1 and M4, m before n at M1 and n before m at M4, then the delivery of m' at M1 will follow the delivery of n' at M1. If n follows m in the total order of messages, then n will be delayed because will know about m. Similarly when m becomes ready at M4 it will know about n and will be delayed if it follows n in the total order of messages.
In the specific ordering displayed above, at M1 n will be delayed when n' is received until m' is received, at which point both m and m are ready and m will be acted on before n. At M4 instead when m' is received it becomes ready and can be applied.

You may find interesting to see in the following diagram all the messages generated for a single message m originated at M2.

The message m becomes ready at M1, .., M4 respectively at a, .., d. Notice that if there are n agents for each original message we will need n-1 multicast messages or (n-1)*(n-1) point to point messages! But all these messages will have the same time stamp as the original m.

We can achieve the same effect of delivering multicast messages in the same order at all sites at the price of using a centralized controller. Now all updates are sent to a single controller, then that controller sends multicasts to all replicated sites. Of course this requires that communication channels are reliable and with the FIFO property.

Consistent Cuts

Important is also the concept of Consistent Cut, that is a cut across the timelines of the computers in the distributed system, such that there is no event a at computer i that is after the cut, and event b at computer j that is before the cut, yet a happened-before b. For example in the diagram below, the first, second, and third cuts are consistent, while the fourth is not.

Thus in a consistent cut we forbid the situation where we see an effect (the arrival of a message) without seeing its cause (the sending of the message). Note that in a consistent cut it is possible to have the sending of a message before a cut, but the receiving of that message after the cut. That means that when considering consistent cuts we have also to worry about and save the 'in transit' messages, i.e. messages sent before the cut but received after the cut.
A consistent cut corresponds to a state in the distributed system which can be explained on the basis of the events that occurred up to that time, there is no evidence in one system of something caused in another system, and the cause has not yet been seen. This is necessary for reasonable debugging and for all situations where we want to be able to observe the evolution of the state of a distributed system. And, most importantly, to create checkpoints for distributed computations.
If in a transaction we transfer $100 from account A to account B, there is a time when the $100 are out of A but not yet in B (or viceversa). The money is like "in transit", or, this is the preferred term, "in the [communication] channel". When we look for consistent cuts (or states), we look for situations where either nothing is in the channel, or where we know what is in the channel.

Chandy and Lamport have an algorithm for computing consistent global states (also called snapshots).

The Chandy-Lamport algorithm allows each node in a distributed system to collect locally information in a way that guaranties that the totality of the information collected represents a consistent global state. But it does not say how to deliver to each (or some) node this total information. This is the Global State Collection Problem. It has the following solution: 

   int i; // The identity of the current node
   State s[i]; // The information collected at node i in Chandy-Lamport
   Set V[i]; // initially V[i] = {s[i]}; It is the info collected at i
   Set W[i]; // initially W[i] is empty; It is the info sent from i

   while (!terminationCondition) {
      if (V[i] != W[i]) then send V[i]-W[i] on all outgoing arcs
      if (arc from j to i contains message M) then V[i] = V[i] union M
   }

   Terminate when V[i]==W[i] for all i and all arcs from any i to any j
   are empty. But how can we determine if these conditions are true?
   We need a termination detection algorithm.
   One such algorithm is the Dijkstra-Scholten algorithm.
And here is the Dijkstra-Scholten Termination Detection Algorithm 

At each node k (we assume that k=0 is the node that starts the algorithm and it has no predecessor) it is run the 
following algorithm:
	Variables:
		C: integer; // Represents the number of messages received by node
			    // and not acknowledged
		D: integer; // Represents the number of messages sent by node 
			    // and not acknowledged
		M: {signal, ack}; // value of a message sent on a link
		sender: integer; // The sender of a message
		parent: integer; // The identity of a node - used to build a 
			    // spanning tree
	begin
		C = D = 0;
		parent = k;
		if (k == 0) {
		    c = 1;
		    for each successor {
			send signal message to that successor;
			D++;
		    }
		}
		while (true) {
		    if (there is a message on a link) {
			set M to the message and sender to its sender;
			if (M == signal) {
			    if (C == 0) {
				C = 1;
				parent = sender;
				for each successor {
				    send signal message to that successor;
				    D++;
		    		}
			    } else if (C == 1) {
				send ack message to sender
			    }
			} else { // The message is an ack
			    D--;
			}
		    }
		    if (C == 1 && D == 0) {
			send ack to parent;
			C = 0;
			parent = k;
			if (k == 0)
			    The algorithm has terminated
		    }
		}
	end
Here is an example:

We show a possible evolution of the algorithm. In the following table we state for each node the value of the parent, C, and D
012345678
0,0,01,0,02,0,03,0,04,0,05,0,06,0,0 7,0,08,0.0
0,1,11,0,02,0,03,0,04,0,05,0,06,0,0 7,0,08,0.0
0,1,10,1,12,0,03,0,04,0,05,0,06,0,0 7,0,08,0.0
0,1,10,1,11,1,23,0,04,0,05,0,06,0,0 7,0,08,0.0
0,1,10,1,11,1,22,1,14,0,05,0,06,0,0 2,1,28,0.0
0,1,10,1,11,1,22,1,13,1,15,0,06,0,0 2,1,17,1,1
0,1,10,1,11,1,22,1,13,1,14,1,18,1,2 2,1,17,1,1
0,1,10,1,11,1,22,1,13,1,15,0,06,0,0 2,1,17,1,0
0,1,10,1,11,1,22,1,14,0,05,0,06,0,0 2,1,08,0,0
0,1,10,1,11,1,03,0,04,0,05,0,06,0,0 7,0,08,0,0
0,1,10,1,02,0,03,0,04,0,05,0,06,0,0 7,0,08,0,0
0,1,01,0,02,0,03,0,04,0,05,0,06,0,0 7,0,08,0,0
0,0,01,0,02,0,03,0,04,0,05,0,06,0,0 7,0,08,0,0

Physical Clocks

Coordinated Universal Time (UTC) is the standard used for time, based on an atomic clock. There are radio stations that transmit a pulse each UTC second. At receivers the accuracy is about 10ms, to account for errors in the transmitter and mutability in atmospheric conditions. Some computers are with UTC receivers and they can transmit time information on the internet to other computers.

Given a computer that has a local physical clock that has drift d (i.e., it will have a maximum error d after one second), if we want it to have an error betweek the clocks of two machines (i.e. the difference, called skew, between the hardware clocks of the two systems at one instant] that is never greater than e, we can do it by requesting a UTC time server for the correct time every e/2d seconds (the factor 2 accounts for the fact that the drift can slow down or speed up the local physical clock). The Christian's Algorithm implements this idea. Here are two potential problems with this algorithm, together with their solution.

A number of other algorithms are used for computing synchronized physical clocks. They are based on various forms of averaging between the times held by various computers and on the use of a number of time servers. The most important time service available is based on the Network Time Protocol (NTP) and involves a number of servers.

The are an unlimited number of uses for synchronized physical clocks and for logical clocks. Among them,