aspbroadcast.java

collective processing environment for mobile devices. It is an environment for mobile device to solv

import java.io.File;
import java.io.FileWriter;
import java.io.PrintWriter;
import java.io.Serializable;

import com.ibm.dthreads.DThread;
import com.ibm.dthreads.Logger;
import com.ibm.dthreads.MP.ObjectMP;

//Application class has to extend DThread or implement DRunnable
public class ASPBroadcast extends DThread{
	
	int[][] data;
	int low;
	int high;
	int size;
	int extra;
	int id;
	int total;
	int noofvertices;
	

      //Constructor that initializes the array

     ASPBroadcast(int noofvertices)
	{
		this.noofvertices = noofvertices;
		data = new int[noofvertices][noofvertices];
		for(int i=0;i<noofvertices;i++)
		{
		  for(int j=0;j<noofvertices;j++)
		  {
			data[i][j]=2;
			if(i==j) data[i][j]=0;
		  }
		} 
	}
	

	//Extends the run() method of the DThread class

	public void run()
	{
	  try
	  {
		long start,end;
		int totalthreads;
		int temp;	
		start=System.currentTimeMillis();
		int rank=getContext().getIdentity();
		totalthreads = getContext().getNumberOfThreads();
		ObjectMP messagepasser = getContext().getMP();
		id=rank;
		int size=noofvertices/totalthreads;
		int extra=(noofvertices%totalthreads);			
		int[] broadcastRow=new int[data.length];
		int bSourceid;
		low=id*size;
		high=low+size;
		int totalRows=data.length;
		if(id==totalthreads-1) 
		
		high=noofvertices;
		System.out.println("Full data length is " + data.length);
		System.out.println("My DThread ID is "+id+"\n My Data Size is "+size+"\n My Lower Limit is "+low+"& Higher Limit is "+high);
		
            //DThread with rank 0 is considered as the Root node for this application
		  if(rank==0)
		  {
			System.out.println("Rank 0 Started");
			for(int k=0;k<data.length;k++)
			{
			 
                   //If kth row is present in this DThread, then broadcast the row to all the other DThreads running 
                   if(k>=low && k<high)
			  {
			  	BroadcastData bData=new BroadcastData(data[k]);
			  	messagepasser.broadcast(bData,Integer.toString(k));
				broadcastRow=data[k];
			  }
                   //If kth row is not present in this DThread, then get the row from the DThread which has the kth row
			  else
			  {
				messagepasser.barrier();
				bSourceid=0;
       			temp=k;
		            while(temp>=size)
			       {
				  bSourceid++;
				  temp-=size;
			        }
			       if(bSourceid>=totalthreads) bSourceid=totalthreads-1;
			       BroadcastData bRow=(BroadcastData)messagepasser.get(bSourceid,Integer.toString(k));
	 			 broadcastRow=bRow.data;
			  }
			  //Computation part
			  for(int i=low;i<high;i++)
			  {
				 for(int j=0;j<data.length;j++)
				 {
					 data[i][j]=Math.min(data[i][j],(data[i][k]+broadcastRow[j]));
				 }
			  }
			}
                  //Create the Output file in the current path
			File output=new File("outputDthread.dat");
			PrintWriter  out=new PrintWriter(new FileWriter(output));

                  //Get the partial result from all the Other Dthreads and compute the final result matrix
			if(totalthreads!=1)
			{
			  messagepasser.barrier();
			  for (int i=1;i<totalthreads;i++)
			  {
			  
			   PartialResult2 iobj = (PartialResult2) messagepasser.get(i,"0");
			   int rowIndex=0;
			   for(int j=iobj.low;j<iobj.high;j++)
				 data[j]=iobj.data[rowIndex++];
				 
			  }
			}

                   //Write the result matrix into the output file
			for(int i=0;i<data.length;i++)
			{
			  for(int j=0;j<data.length;j++)
			  {
				out.println(new Integer(data[i][j]).toString());
			  }
			}
			out.close();
			
			System.out.println("Please see outputDthread.dat for the result");
		 }
			 
             //If the DThread is not root
 		 else
		  {
		 System.out.println("Full data length is " + data.length);

		 for(int k=0;k<data.length;k++)
		 {

               //If kth row is present in this DThread, then broadcast the row to all the other DThreads running 
		   if(k>=low && k<high)
		   {
			 BroadcastData bData=new BroadcastData(data[k]);
			 messagepasser.broadcast(bData,Integer.toString(k));
			broadcastRow=data[k];
		   }

               //If kth row is not present in this DThread, then get the row from the DThread which has the kth row
		   else
		   {
			 messagepasser.barrier();
			  bSourceid=0;
			 temp=k;
         		 while(temp>=size)
			 {
				bSourceid++;
			    temp-=size;
			 }
			 if(bSourceid>=totalthreads) bSourceid=totalthreads-1;
			 BroadcastData bRow=(BroadcastData)messagepasser.get(bSourceid,Integer.toString(k));
			 broadcastRow=bRow.data;
		   }
			  
		   for(int i=low;i<high;i++)
		   {
			  for(int j=0;j<data.length;j++)
			  {
				  data[i][j]=Math.min(data[i][j],(data[i][k]+broadcastRow[j]));
			  }
		   }
		 }

             //Build the Partial result and send to the root node ( in my case it is DThread with rank 0)
             
		  PartialResult2 presult=new PartialResult2(data,low,high);
		  messagepasser.put(presult,rank,0,"0");
		  messagepasser.barrier();
        	  }         
		  System.out.println(" I with id "+id+"Completed");
		  end=System.currentTimeMillis();
		  System.out.println("Time taken is  "+(end-start)+"milliSeconds");
}

catch(Exception e)
{
  System.out.println("Exception ::" + e.getMessage());
  e.printStackTrace();
}
	
}
}


//Class which will hold the partial result
class PartialResult2 implements Serializable
{
	int[][] data;
	int row,col;
	int low;
	int high;
	PartialResult2(int[][] data,int low,int high)
	{
		row=high-low;
		col=data.length;
		this.data=new int[row][col];
		for(int i=low,m=0;i<high;i++,m++)
		{
		  for(int j=0,n=0;j<col;j++,n++)
		  { 
			 this.data[m][n]=data[i][j];
		  }
		}
		this.low=low;
		this.high=high;
	}
}


//Class to have the Row to be broadcasted
 class BroadcastData implements Serializable
{
	 int[] data;
	 
	BroadcastData(int[] data)
	 {
		this.data = data;
	 }
}