Design & implementation of (totally ordered multicast) using lamport logical clocks(Synchronization)

This post deals with the various issues and design decisions  faced during implementation of synchronization between processes using Lamport Logical clocks in a distributed system.

Lamport Logical clock concept can be used implement synchronization between multiple processes ,where each process observe the changes in the same order ,regardless of the order in which the updates originated .

Below are few lines from wikipedia which gives a little background on Lamport’s logical clocks

The algorithm of Lamport timestamps is a simple algorithm used to determine the order of events in a distributed computer system. As different nodes or processes will typically not be perfectly synchronized, this algorithm is used to provide a partial ordering of events with minimal overhead, and conceptually provide a starting point for the more advanced Vector clockmethod.

Distributed algorithms such as resource synchronization often depend on some method of ordering events to function. For example, consider a system with two processes and a disk. The processes send messages to each other, and also send messages to the disk requesting access. The disk grants access in the order the messages were sent. Now, imagine process 1 sends a message to the disk asking for access to write, and then sends a message to process 2 asking it to read. Process 2 receives the message, and as a result sends its own message to the disk. Now, due to some timing delay, the disk receives both messages at the same time: how does it determine which message happened-before the other? (Ahappens-before B if one can get from A to B by a sequence of moves of two types: moving forward while remaining in the same process, and following a message from its sending to its reception.) A logical clock algorithm provides a mechanism to determine facts about the order of such events.

Lamport Requirements (aka “The Requirement”)

Posted above is the link which has the formal  requirements of the system to be designed.

So we basically we have 3 processes running on different systems.Each of the processes generate  X and Y co-ordinates at random intervals which is multicast to other participating processes including itself.The challenge is to see that each of  processes should  see the updates of X and Y co-ordinates of self and other processes  in the same order regardless of when the actual update was issued.

Lets call the three processes P0,P1,P2.  First phase is connection phase.

Connection phase uses TCP protocol to establish connection between the processes. Higher order process connects with lower order process.order is determined by process number.

Po is connected by P1 and P2.

P1 is connected by P2

P0 opens up two server ports for establishing connection with P1 and P2. P1 and P2 connects to P0 as clients.

P1 opens up opens one server port for connecting with P2 and also connects to P0 as client.

P2 opens up 2 client ports and connects with P1 and P0.

And that completes connection phase.

Logical clock Requirement.

You need to have a logical clock on each of these processes and it is required that no two processes have same clock value at any instant.This uniqueness in clock value is achieved by appending process ID to the clock value.

For example, Lets say if process 1 and process 2 both have clock value of 40.We are avoiding such a situation by appending process id to the clock.  So ,In this case we would have the time of process 1 and process 2 as 40.1 and 40.2 respectively.

Also the requirement specifies each clock has different clock rate. i.e each clock increments at its own specified rate for every second.

Process 0  :clock rate : 4

seconds    clock

1                   4.0  [clock value   .  process number ]

2                   8.0

3                   12.0

Programming Tip:

I initially declared clock as  ”double” to accommodate point value.But i figured out that double addition is not consistent.you sometimes end up having values like  11.999999 instead of 12.this is undesirable.So i decided to consider lowest unit possible.Also i believe same concept is used for counting money.so you keep cents as unit of money  instead of dollars ,that way you would not have to deal with inconsistency in double or floating points.

Also i initialized clock to their process ID instead of default 0.that way none of the clocks would have same time ever. Needless to say,clock executes in a separate thread.

Running phase

In Running phase, three processes generates updates at a random interval.Once processes generate an update, a message is composed comprising of  payload which contains generated X and Y co-ordinates,time stamp of message and process id of the sender.This UPDATE message is multicast to all participating processes including itself.Once participating process receives an UPDATE message, the message is put on to their Queue(“UPDATE QUEUE”) maintained at each process, and an ACK is multicast to all other processes.ACK contains the timestamp of the UPDATE message.

Once  a process receives 3 ACK messages it can remove the corresponding UPDATE message from UPDATE QUEUE and commit the changes .

Implementation Specifics:

You can declare UPDATE QUEUE as static and synchronize the operation on this QUEUE by  making the method which modifies the Queue “synchronized”.Another Alternative is you can use a Vector instead of Queue which is synchronized by default.

Once the processes are connected to each other,you can call a method to initialize all the required ObjectOutputStream Objects and ObjectInputStream Objects which are used for receiving Messages as well as Multicast Message.previously i used to call “new” on ObjectOutputStream and ObjectInputStream every time i needed to write a Message or read a Message.

I realized this was in-efficient and not necessary.So i decided to initialize ObjectOutputStream and ObjectInputStream  objects once connection between all processes were established.Also i learnt that ObjectOutputStream should be initialized first followed by ObjectInputStream. That is because if you are transmitting data in a bi-directional fashion  (2 -ways) then, the input stream constructor waits for an initial string of bytes to arrive from the ObjectOutputStream open().In bi-directional ,both ends are waiting for construction of ObjectOutputStream and neither of them can proceed.

I also had a single Receiver thread which would repeatedly read objects from different processes in sequential fashion.I thought this was ineffective and spawned separate exclusive Receiver thread for each process.Things worked smoothly and now I had multicast working properly

The problem i faced was in handling ACKs received from other processes.Every time i received an ACK i would map it to Message in UPDATE Queue and increment the ACK Counter.But this model assumed corresponding ACK for an UPDATE Message always occurred after UPDATE Message was received which was supposed to happen in ideal case.But there were scenarios where i received an ACK for an UPDATE which had not yet arrived. Visualizing this scenario ate some precious time.

To handle this situation ,intuitively ,i first thought of collecting all ACKS  which did not map to any UPDATE Message in container(Vector or Queue) and then handle these ACKs when process received next UPDATE message.Although this improved the situation by marginal amount ,it failed miserably in preserving the total order.

During debugging phase ,i observed that Messages populated in UPDATE Queue in totally ordered fashion for all the processes and that was a good thing.The only requirement was to handle ACKs generated by processes to remove UPDATE messages and commit them.

Previous approaches had tightly coupled interaction between ACKs and UPDATE messages.The following approach loosened it a bit and finally worked smoothly.

To handle ACKs received from other processes ,i created a “static” hashmap with timestamp as key(as it is always unique) and number of occurences of ACK as  value.As and when ACKS were received i populated them into a hashmap incremented its occurrence.When once occurrence of an ACK changed to 3,i  removed an UPDATE Message from Message QUEUE and committed the changes.This worked perfectly and preserved total ordering as expected.

Finishing Phase:

In finishing phase ,each process sends a “FINISH”  Message to other processes and indicates that it is done with all the updates. the process terminates when it receives required number of “FINISH” Messages.

Update:

I have added the source code in my github account.

https://github.com/sreenidhibs/Lamport-Time-Clock—Virtual-Sychrony—Sequential-Consistency