Lamport's logical clocks
Method: The logical clock runs on top of some other message-passing protocol, adding additional state at each process and additional content to the messages (which is invisible to the underlying protocol). Every process maintains a local logical clock. When a process sends a message or executes and internal step, it sets clock ? clock + 1. If it sends a message, it piggybacks the resulting clock value on the message. When a process receives a message, it sets clock ? max(clock, message timestamp)+1; the resulting clock value is taken as the time of receipt of the message.
Claim: If we order all events by clock value, we get an execution of the underlying protocol that is locally indistinguishable from the original execution.
Proof: Key observation is that clock-updating rules ensure (a) all events at the same process occur in increasing order and (b) the receipt of a message has a higher clock value than its transmission. In other words the clock order is consistent with the causal order. So we can apply the local indistinguishability theorem from Synchronizers to argue that the reordered execution does what we want.
Logical Clocks
A logical clock C is a map from the set of events E to N (the set of natural numbers) with the following constraint:
/*program to implement lamports logical clock*/
#include
#include
#include
void main()
{
int n,i,a[100],m,j,c[100][100],d[100],temp,s[100][100],g,r[100][100];
clrscr();
printf("\nenter the no. of process in the distributed system=");
scanf("%d",&n);
for(i=0;i
scanf("%d",&a[i]);
}
for(i=0;i
scanf("%d",&d[i]);
}
printf("\the overall configuration of the distributed system is");
for(i=0;i
printf("\nthe process P[%d] has %d events in it and its drift rate is %d",i,a[i],d[i]);
}
//for each process, entering happened before relationship of events within that process
for(j=0;j
for(i=0;i
printf("\nenter the timestamp of %d event of process P[%d]",i+1,j);
scanf("%d",&c[j][i]);
}
}
//computing happened before reln. among the events of single process
for(j=0;j
for(i=0;i
if(c[j][i]>c[j][i+1])
{
temp=c[j][i];
c[j][i]=c[j][i+1];
c[j][i+1]=temp;
}
}
}
for(j=0;j
printf("\nthe non-decreasing order of events of a process P[%d] are",j);
for(i=0;i
printf("\nthe c[%d][%d] event with timestamp value %d is %d in the process",j,i,c[j][i],i);
}
}
//considering the inter-process communication messages
printf("\nenter the no. of sending events=");
scanf("%d",&m);
for(g=0;g
printf("\nenter the timestamp index of send message event=");
scanf("%d\t%d",&s[j][i]);
s[j][i]=c[j][i];
}
for(g=0;g
printf("\nthe sending events in the distributed system are s[%d][%d] with timestamp values %d",j,i,s[j][i]);
}
//establishing happened before relationship among the events of the processes
//using the implementation rule 1 and 2
c[j][i]=c[j][i]+d[j];//incrementing the values
//for IPC
c[j][i]=max(c[j][i]
getch();
}
II-----------
/*
A basic extension of the java.applet.Applet class
*/
import java.applet.*;
import java.awt.*;
import java.io.*;
import java.net.*;
import java.util.*;
public class chatServer extends Applet
{
client clientObject;
String line;
public void init()
{
super.init();
setLayout(null);
addNotify();
resize(640,480);
txtName = new java.awt.TextField();
txtName.setText("ID");
txtName.reshape(72,12,168,27);
txtName.setFont(new Font("Dialog", Font.BOLD, 14));
txtName.setBackground(new Color(16776960));
add(txtName);
lblIp = new java.awt.Label("IP Address");
lblIp.reshape(240,12,96,24);
lblIp.setFont(new Font("Dialog", Font.BOLD, 14));
add(lblIp);
lblPort = new java.awt.Label("Port");
lblPort.reshape(480,12,48,24);
lblPort.setFont(new Font("Dialog", Font.BOLD, 14));
add(lblPort);
txtPort = new java.awt.TextField();
txtPort.setText("4321");
txtPort.reshape(528,12,72,27);
txtPort.setFont(new Font("Dialog", Font.BOLD, 14));
txtPort.setBackground(new Color(16776960));
add(txtPort);
txtArea = new java.awt.TextArea();
txtArea.setEditable(false);
txtArea.reshape(24,72,588,288);
txtArea.setBackground(new Color(65535));
add(txtArea);
btnConnect = new java.awt.Button("Connect");
btnConnect.reshape(276,432,84,34);
add(btnConnect);
txtField = new java.awt.TextField();
txtField.reshape(24,384,587,32);
txtField.setFont(new Font("Dialog", Font.BOLD, 14));
txtField.setBackground(new Color(16776960));
add(txtField);
txtIPaddr = new java.awt.TextField();
txtIPaddr.setText("161.178.121.1");
txtIPaddr.reshape(336,12,144,27);
txtIPaddr.setFont(new Font("Dialog", Font.BOLD, 14));
txtIPaddr.setBackground(new Color(16776960));
add(txtIPaddr);
label1 = new java.awt.Label("Enter your ID and press connect to begin");
label1.reshape(156,48,312,19);
label1.setFont(new Font("Dialog", Font.BOLD, 14));
label1.setBackground(new Color(12632256));
add(label1);
lblName = new java.awt.Label("Name");
lblName.reshape(12,12,60,24);
lblName.setFont(new Font("Dialog", Font.BOLD, 14));
add(lblName);
}
public boolean action(Event event, Object arg)
{
if (event.target == btnConnect)
{
btnConnect.disable();
txtField.requestFocus();
clientObject = new client(this, txtIPaddr.getText(), txtPort.getText());
return true;
}
if (event.target == txtField)
{
line = txtField.getText();
txtField.setText("");
String lineOut = "00000000|" + txtName.getText() + "|" + line + "|";
clientObject.pout.println(lineOut);
return true;
}
return super.action(event, arg);
}
java.awt.TextField txtName;
java.awt.Label lblIp;
java.awt.Label lblPort;
java.awt.TextField txtPort;
java.awt.TextArea txtArea;
java.awt.Button btnConnect;
java.awt.TextField txtField;
java.awt.TextField txtIPaddr;
java.awt.Label label1;
java.awt.Label lblName;
}
class client extends Thread {
DataInputStream dis;
Socket sock;
PrintStream pout;
InputStream in;
OutputStream out;
DataInputStream din;
DataOutputStream dout;
String IPA;
chatServer myClient;
String port;
int portInt;
client(chatServer myClient, String IPA, String port) {
this.myClient = myClient;
this.IPA = IPA;
this.port = port;
portInt = Integer.valueOf(port).intValue();
start();
}
public void run() {
System.out.println("client thread started");
try {
sock = new Socket(IPA, portInt);
System.out.println("connection made");
in = sock.getInputStream();
out = sock.getOutputStream();
pout = new PrintStream(out);
din = new DataInputStream(in);
dout = new DataOutputStream(out);
while(true) {
String request = din.readLine();
myClient.txtArea.appendText(request + "\n");
}
}
catch (UnknownHostException e ) {System.out.println("can't find host"); }
catch ( IOException e ) {System.out.println("Error connecting to host");}
}
}
III---------------------------
// RMI
import java.math.*;
import java.rmi.*;
import java.rmi.server.*;
public class PowerServiceServer extends UnicastRemoteObject
implements PowerService
{
public PowerServiceServer () throws RemoteException
{
super();
}
public BigInteger square ( int number )
throws RemoteException
{
String numrep = String.valueOf(number);
BigInteger bi = new BigInteger (numrep);
bi.multiply(bi);
return (bi);
}
public BigInteger power ( int num1, int num2)
throws RemoteException
{
String numrep = String.valueOf(num1);
BigInteger bi = new BigInteger (numrep);
bi = bi.pow(num2);
return bi;
}
public static void main ( String args[] ) throws Exception
{
if (System.getSecurityManager() == null)
System.setSecurityManager ( new RMISecurityManager() );
PowerServiceServer svr = new PowerServiceServer();
Naming.bind ("PowerService", svr);
System.out.println ("Service bound....");
}
}
IIII---------------------
Introduction:
If a collection of processes share a resource or collectionof resources, then often mutual exclusion is required to revent interferences and ensure consistency when accessing the resources. This is critical section problem, familiar in the domain of operating systems. In a distributed system, however, neither sahred variables nor facilites supplied by a single local kernel can be used to solve it, in general. We require a solution to distributed mutual exclusion: one that is based solely on message passing.
Algorithm:
On initiation
state:= RELEASED;
To enter the section
state:= WANTED;
Multicast request to all processes;
T:= request's timestamp;
Wait until (number of replies receeived = (N-1));
state:= HELD;
On receipt of a request
if(state=HELD or (state = WANTED and (T,pj)<(Ti,pi)))
Then
Queue request from pi without replying;
Else
Reply immediately to pi;
End if
To exit the ciritical section
State:= RELEASED;
Reply to any queued request;
package MutualExclusion;
import daj.*;
import java.util.*;
public class RicartAgrawala extends Application {
public static final int PNUM = 6;
public static final int DISTANCE = 70;
public static final int BORDER = 40;
public static void main(String[] args) {
new RicartAgrawala().run();
}
/** Creates new Main */
public RicartAgrawala() {
super("RicartAgrawala Mutual Exclusion",
(int) ((PNUM * DISTANCE) / Math.PI) + 2 * BORDER,
(int) ((PNUM * DISTANCE) / Math.PI) + 2 * BORDER);
}
public void construct() {
Node[] nodes = new Node[PNUM];
int i, j;
double phi, r = (PNUM * DISTANCE) / (2 * Math.PI);
for (i=0; i
nodes[i] = node(new Prog(i), new Integer(i).toString(),
(int) (r * Math.cos(phi) + r + BORDER),
(int) (r * Math.sin(phi) + r + BORDER));
}
for (i=0; i
//link(nodes[j], nodes[i]);
}
}
}
public String getText() {
return "An implementation of the RicartAgrawala Algorithm for\n" +
"solving the distributed mutual exclusion problem.";
}
}
/**
* The code that will be executed on each node.
* @author rene
* @version
*/
class Prog extends Program {
/** possible value for region: idle */
public static final int R = 0;
/** possible value for region: try to get the ressource */
public static final int T = 1;
/** possible value for region: critical region entered */
public static final int C = 2;
/** possible value for region: exiting critical region */
//public static final int E = 3;
/** the node number */
protected int index;
/** the region can be either R, T, C or E */
protected int region = R;
/** the current clock value */
protected int clock = 0;
/** for every node, the message history */
protected Vector[] history = null;
/** for every neighbour, the buffer for remembering deferred ok messages
until critical region has been left */
protected Vector[] deferred_ok = null;
/** the logical time of the last "try" message that has been sent */
protected LogicalTime lastTryTime = null;
/** the default constructor initializes the index and the arrays */
public Prog(int index) {
this.index = index;
history = new Vector[RicartAgrawala.PNUM];
for (int i=0; i
deferred_ok = new Vector[RicartAgrawala.PNUM];
for (int i=0; i
deferred_ok[i] = new Vector();
else
deferred_ok[i] = null;
}
/** the user function for trying to get the ressource (getting the
permission to enter the critical region) */
public void tryRessource() {
// single actions must be atomic --> synchronize them
synchronized (this) {
clock++;
region = T;
lastTryTime = new LogicalTime(clock, index);
out().send(new Msg(Msg.TRY, lastTryTime));
}
}
/** the user function for freeing the ressource (exiting the critical
region) */
public void freeRessource() {
synchronized (this) {
clock++;
// this does not match the formal specification - the release is
// done immediately
region = R;
for (int i=0; i
for (int j=0; j
}
deferred_ok[i].removeAllElements();
}
}
}
/** this function is called when the ressource can be taken (the critical
region can be entered) */
public void ressourceAvailable() {
// for testing purposes there is no user to ne notified
}
/** this function is called when the ressource has been freed (the critical
region has been left) */
public void ressourceFreed() {
// for testing purposes there is no user to ne notified
}
/** this method performs the node's main functions */
public void main() {
Random rand = new Random();
GlobalAssertion safety = new CriticalRegionConflict(),
liveness = new CriticalRegionLockout();
while (true) {
assert(safety);
assert(liveness);
/* first the tree construction routines */
int received = in().select(2);
if (received >= 0) {
synchronized (this) {
Msg msg = (Msg) in(received).receive();
if (msg.time.clock > clock)
clock = msg.time.clock;
clock++;
history[received].add(msg);
// respond to try requests from other nodes
if (msg.type == Msg.TRY && received != index)
handleTryMessage(received, msg);
// check if the critical region can be entered now
if (region == T && checkOkMsgsReceived()) {
clock++;
region = C;
// the last try is invalid now
lastTryTime = null;
}
}
}
// simulate the user here
if (region == R && rand.nextInt(RicartAgrawala.PNUM * 2) == 0) {
tryRessource();
}
else if (region == C && rand.nextInt(RicartAgrawala.PNUM) == 0) {
freeRessource();
}
}
}
public String getText() {
String s = "Node " + index + "\nRegion: ";
switch (region) {
case R: s += "idle";
break;
case T: s += "trying to get lock:\n";
s += "try time = " + lastTryTime + "\n";
break;
case C: s += "inside critical region";
break;
/*case E: s += "exiting critical region";
break;*/
default: s += "invalid";
}
s += "\nClock value: " + clock;
return s;
}
/** This is a helper function that responds properly to a received "try"
message. */
private void handleTryMessage(int received, Msg msg) {
Msg ackMsg = new Msg(Msg.OK, new LogicalTime(clock, index));
if (region == R /*|| region == E*/)
out(received).send(ackMsg);
else if (region == C)
deferred_ok[received].add(ackMsg);
else if (region == T) {
if (lastTryTime != null && msg.time.lessThan(lastTryTime))
out(received).send(ackMsg);
else
deferred_ok[received].add(ackMsg);
}
}
/** This is a helper function that checks if all "OK" messages have been
recieved for the last try message. */
private boolean checkOkMsgsReceived() {
boolean receivedOkMsgs = true;
for (int i=0; i
// always check the last "OK" msg of the history
// queues, because this is the one that has been
// recieved at last
int j = history[i].size() - 1;
while (j >= 0 && ((Msg) history[i].get(j)).type != Msg.OK)
j--;
if (j >= 0) {
Msg lastMsg = (Msg) history[i].get(j);
if (! lastTryTime.lessThan(lastMsg.time))
// there is no "OK" message that is newer
// than our own "try" message
receivedOkMsgs = false;
}
else
// there is no "OK" message from this node
receivedOkMsgs = false;
}
}
return receivedOkMsgs;
}
}
/** Represents a logical time value with operations for checking for equality
* and order.
*/
class LogicalTime {
/** the clock value */
public int clock;
/** for equal clock values the sender process's index is compared */
public int process;
/** a logical time object can only be constructed when both parameters are
known */
public LogicalTime(int clock, int process) {
this.clock = clock;
this.process = process;
}
/** checks for equality as defined for the Lamport logical time */
public boolean equals(Object o) {
if (o instanceof LogicalTime &&
((LogicalTime) o).clock == clock &&
((LogicalTime) o).process == process)
return true;
else
return false;
}
/** check if the the current time is less than t as defined for the Lamport
logical time */
public boolean lessThan(LogicalTime t) {
if (clock < t.clock ||
(clock == t.clock && process < t.process))
return true;
else
return false;
}
public String toString() {
return "(" + clock + ", " + process + ")";
}
}
class Msg extends Message {
/** message type: try (a process tries to get the shared ressource) */
public static final int TRY = 0;
/** message type: ok (a process confirms another process's request */
public static final int OK = 1;
/** the message type */
public int type;
/** the logical time of the sending process when the message was sent */
public LogicalTime time;
public Msg(int type, LogicalTime time) {
this.type = type;
this.time = time;
}
public boolean equals(Object o) {
if (o instanceof Msg &&
((Msg) o).type == type &&
((Msg) o).time.equals(time))
return true;
else
return false;
}
public String getText() {
String s = "Message type: ";
switch (type) {
case TRY: s += "try";
break;
case OK: s += "ok";
break;
default: s += "invalid";
}
s += "\nTime: " + time.toString();
return s;
}
}
/** This assertion checks that at most 1 process is in the critical region at
the same time. */
class CriticalRegionConflict extends GlobalAssertion {
private Vector procsInCR = new Vector();
public boolean assert(Program progs[]) {
int numOfProcsInCR = 0;
procsInCR.removeAllElements();
for (int i=0; i
numOfProcsInCR++;
procsInCR.add(new Integer(i));
}
}
return numOfProcsInCR <= 1;
}
public String getText() {
String text = "Processes in the critical region: ";
for (int i=0; i
return text;
}
}
/** This assertion checks that if a process tries to enter the critical region,
no other process trying it at any time later will be allowed to enter the
critical region before the first process has entered it. */
class CriticalRegionLockout extends GlobalAssertion {
private LogicalTime[] tryTimes = new LogicalTime[RicartAgrawala.PNUM];
private int procInCR = -1, procTryingLonger = -1;
public CriticalRegionLockout() {
for (int i=0; i
}
public boolean assert(Program progs[]) {
// remember the times when the processes tried to enter the CR
for (int i=0; i
tryTimes[i] = ((Prog) progs[i]).lastTryTime;
// Now check when a process is in the CR, if another one is still
// trying but started to try earlier. This should not happen.
for (int i=0; i
for (int j=0; j
tryTimes[j].lessThan(tryTimes[i])) {
procInCR = i;
procTryingLonger = j;
return false;
}
}
}
return true;
}
public String getText() {
return "Process " + procInCR + " is in the critical region, but " +
procTryingLonger + "was trying earlier !";
}
}
