Two-Phase Commit Protocol

Two-Phase Commit Protocol

PROBLEM STATEMENT

Write a program to implement a two-phase commit protocol. Messages are passed between ‘n’ nodes (or terminals) and actions are committed or aborted, with respect to the response of other nodes. The final action is the n communicated to all other nodes.

DESCRIPTION

Two-Phase Commit Protocol follows following steps;

Step1:
The Coordinator sends a VOTE_REQUEST message to all participants.

Step2:
When a participant receives a VOTE_REQUEST message,it returns either a VOTE_COMMIT message to the coordinator telling the coordinator that is prepared to commit or a VOTE_ABORT message.

Step3:
The coordinator collects all votes from the participants. If all participants had voted to commit the transaction then it sends GLOBAL_COMMIT message to all participants. Even if one participant had voted to abort the transaction then it send GLOBAL_ABORT message.

Step4:
Each participant that voted for a commit waits for the final reaction by the coordinator. If a participant receives a GLOBAL_COMMIT message it locally commits the transaction else it abort the local transaction.

      
SOURCE CODE:

TPCServer.java

import java.io.*;
import java.net.*;


public class TPCServer
{

    public static void main(String a[])throws Exception
        {
            BufferedReader br;
            InetAddress lclhost;
            lclhost=InetAddress.getLocalHost();
            Server ser=new Server(lclhost);
           
            System.out.println("Server in sending mode.....");
           

            // Sending data to client 1
            ser.setSendPort(9000);   //recport=8000
            ser.setRecPort(8001);   //sendport=9001

            System.out.println("Send request data to client1..");
        br=new BufferedReader(new InputStreamReader(System.in));
        String s=br.readLine();
            System.out.println("Data is "+s);
            ser.sendData();
            System.out.println("Waiting for response from client1....");
            ser.recData();
   
            // Sending data to client 2
            ser.setSendPort(9002);  //recport=8002
            ser.setRecPort(8003);  //senport=9003
            System.out.println("Send request data to client2..");
        br=new BufferedReader(new InputStreamReader(System.in));
        String s1=br.readLine();
            System.out.println("Data is "+s1);
            ser.sendData();
            System.out.println("Waiting for response from client2....");
            ser.recData();
           

            //Sending the final result to client 1
            ser.setSendPort(9000);
            ser.sendData();           

            //Sending the final result to client 2
            ser.setSendPort(9002);
            ser.sendData();       
        }
}


class Server
{

    InetAddress lclhost;
    int sendPort,recPort;
    int ssend =0;
    int scounter=0;
    Server(InetAddress lclhost)
    {
        this.lclhost=lclhost;
    }

    public void setSendPort(int sendPort)
    {
        this.sendPort=sendPort;
    }
   
    public void setRecPort(int recPort)
    {
        this.recPort=recPort;
    }

    public void sendData()throws Exception
    {
       
        DatagramSocket ds;
        DatagramPacket dp;
        String data="";

        if(scounter<2 && ssend<2)
            {
                data="VOTE_REQUEST";
            }

        if(scounter<2 && ssend>1)
            {
                data="GLOBAL_ABORT";
                data= data + " TRANSACTION ABORTED";
               
            }
        if(scounter==2 && ssend>1)
            {
                data="GLOBAL_COMMIT";
                data= data + " TRANSACTION COMMITED";

           
            }
       
       
       
        ds=new DatagramSocket(sendPort);
        dp=new DatagramPacket(data.getBytes(),data.length(),lclhost,sendPort-1000);
        ds.send(dp);
        ds.close();
        ssend++;
    }


public void recData()throws Exception
        {
            byte buf[]=new byte[256];
            DatagramPacket dp=null;
            DatagramSocket ds=null;
            String msgStr="";
            try{
            ds=new DatagramSocket(recPort);
            dp=new DatagramPacket(buf,buf.length);
            ds.receive(dp);
            ds.close();
            }
            catch(Exception e)
            {
                e.printStackTrace();
            }
             msgStr=new String(dp.getData(),0,dp.getLength());
System.out.println("String = "+msgStr);           
if(msgStr.equalsIgnoreCase("VOTE_COMMIT"))
                {
                    scounter++;
                }
        System.out.println("Counter value = "+scounter + "\n Send value = "+ssend);
    }       

};

TPCClient1.java

import java.io.*;
import java.net.*;


public class TPCClient1
{

    public static void main(String a[])throws Exception
        {
            InetAddress lclhost;
            lclhost=InetAddress.getLocalHost();
            Client clnt=new Client(lclhost);

           
            clnt.setSendPort(9001);   //recport=8000
            clnt.setRecPort(8000);   //sendport=9001
            clnt.recData();
            clnt.sendData();
            clnt.recData();
        }
}


class Client
{
InetAddress lclhost;
    int sendPort,recPort;
    Client(InetAddress lclhost)
    {
        this.lclhost=lclhost;
}

    public void setSendPort(int sendPort)
    {
        this.sendPort=sendPort;
    }
   
    public void setRecPort(int recPort)
    {
        this.recPort=recPort;
    }

    public void sendData()throws Exception
    {
        BufferedReader br;
        DatagramSocket ds;
        DatagramPacket dp;
        String data="";
        System.out.println("Enter the Response 'VOTE_COMMIT' || 'VOTE_ABORT' ");
        br=new BufferedReader(new InputStreamReader(System.in));
        data = br.readLine();
        System.out.println("Data is "+data);

       
        ds=new DatagramSocket(sendPort);
        dp=new DatagramPacket(data.getBytes(),data.length(),lclhost,sendPort-1000);
        ds.send(dp);
        ds.close();
       
    }


public void recData()throws Exception
        {
            byte buf[]=new byte[256];
            DatagramPacket dp;
            DatagramSocket ds;
           
            ds=new DatagramSocket(recPort);
            dp=new DatagramPacket(buf,buf.length);
            ds.receive(dp);
            ds.close();
            String msgStr=new String(dp.getData(),0,dp.getLength());
            System.out.println("Client1 data " +msgStr);
           
    }       

};

TPCClient2.java

import java.io.*;
import java.net.*;


public class TPCClient2
{

    public static void main(String a[])throws Exception
        {
            InetAddress lclhost;
            lclhost=InetAddress.getLocalHost();
            Client client=new Client(lclhost);

            // Sending data to client 2
            client.setSendPort(9003);  //recport=8002
            client.setRecPort(8002);  //senport=9003
            client.recData();
            client.sendData();
            client.recData();
           
    }
}


class Client
{

    InetAddress lclhost;
    int sendPort,recPort;
    Client(InetAddress lclhost)
    {
        this.lclhost=lclhost;
    }

    public void setSendPort(int sendPort)
    {
        this.sendPort=sendPort;
    }
   
    public void setRecPort(int recPort)
    {
        this.recPort=recPort;
    }

    public void sendData()throws Exception
    {
        BufferedReader br;
        DatagramSocket ds;
        DatagramPacket dp;
        String data="";
        System.out.println("Enter the Response 'VOTE_COMMIT' || 'VOTE_ABORT' ");
        br=new BufferedReader(new InputStreamReader(System.in));
       
        data = br.readLine();
        System.out.println("Data is "+data);

        ds=new DatagramSocket(sendPort);
        dp=new DatagramPacket(data.getBytes(),data.length(),lclhost,sendPort-1000);
        ds.send(dp);
        ds.close();
       
    }

public void recData()throws Exception
        {
            byte buf[]=new byte[256];
            DatagramPacket dp;
            DatagramSocket ds;
            ds=new DatagramSocket(recPort);
            dp=new DatagramPacket(buf,buf.length);
            ds.receive(dp);
            ds.close();
            String msgStr=new String(dp.getData(),0,dp.getLength());

            System.out.println(msgStr);           
    }       

};

OUTPUT

TPCServer.java

C:\Documents and Settings\Nids\Desktop\DCpracs\twoPhaseCommit_prac7>java TPCServ
er
Server in sending mode.....
Send request data to client1..
VOTE_REQUEST
Data is VOTE_REQUEST
Waiting for response from client1....
String = VORE_COMMIT
Counter value = 0
 Send value = 1
Send request data to client2..
VOTE_REQUEST
Data is VOTE_REQUEST
Waiting for response from client2....
String = VOTE_COMMIT
Counter value = 1
 Send value = 2

TPCClient1.java

C:\Documents and Settings\Nids\Desktop\DCpracs\twoPhaseCommit_prac7>java TPCClie
nt1
VOTE_REQUEST
Enter the Response 'VOTE_COMMIT' || 'VOTE_ABORT'
VORE_COMMIT
Data is VORE_COMMIT
GLOBAL_ABORT TRANSACTION ABORTED

TPCClient2.java

C:\Documents and Settings\Nids\Desktop\DCpracs\twoPhaseCommit_prac7>java TPCClie
nt2
VOTE_REQUEST
Enter the Response 'VOTE_COMMIT' || 'VOTE_ABORT'
VOTE_COMMIT
Data is VOTE_COMMIT
GLOBAL_ABORT TRANSACTION ABORTED