application.java

发布/订阅系统路由重配算法,可应用于ad hoc环境

package sim;

import broker.*;
import java.util.*;

public class Application {
	
	/**
	 * Constant to mark a constant time interval.
	 */
	protected static final int CONSTANT = 0;
	
	/**
	 * Constant to mark an exponential distribution.
	 */
	protected static final int EXPONENTIAL = 1;
	
	/**
	 * The number of active jobs. A job consits of one publisher and
	 * several subscribers interested in his notifications.
	 */
	protected static int numberOfJobs = 50;
	
	/**
	 * The number of subscribers of a job. These subscribers are interested
	 * in the notifications of the job's publisher. 
	 */
	protected static int numberOfSubscribers = 9;
	
	/**
	 * The interval type between two consecutive notifications of a job.
	 * The interval can be "CONSTANT" or "EXPONENTIAL" distributed.
	 */
	protected static int publicationInterval = EXPONENTIAL;

	/**
	 * Constant time interval or 'lambda' of an exponential distribution.
	 */
	//protected static double publicationConstant = 2;
	protected static double publicationConstant = 0.2;
	
	/**
	 * The interval between a reassignment of jobs to brokers. 
	 * By reassigning jobs a change of the load may be simulated.
	 */
	protected static int reassignmentInterval = CONSTANT;
	
	/**
	 * Constant time interval or 'lambda' of an exponential distribution.
	 */
	protected static double reassignmentConstant = 5001d;
	
	/**
	 * Fraction of locality considered by assigning jobs to brokers.
	 * In simulations with a high locality the subscribers are placed
	 * close to the job's publisher.
	 */
	protected static double locality = 0.0d;
	
	public static double localityPower = 4.0d;
	
	/**
	 * Random seed for all random values concerning the application.
	 */
	protected static long applicationSeed = 7390923261026300538L;
	
	/**
	 * The random generator initialized with the given applicationSeed.
	 */
	protected static Random rand = new Random(applicationSeed);
	
	
	/**
	 * Initializes all fields of the application class.
	 * The values are stored as Strings in the properties object.
	 * The keys are the names of the fields. 
	 * @param properties properties containing name/value pairs.
	 */
	public static void initializeProperties(Properties properties){
		// number of jobs
		numberOfJobs = properties.getProperty("numberOfJobs") != null ?
				       Integer.parseInt(properties.getProperty("numberOfJobs")) :
                       numberOfJobs;
	    // number of subscribers
        numberOfSubscribers = properties.getProperty("numberOfSubscribers") != null ?
        		              Integer.parseInt(properties.getProperty("numberOfSubscribers")) :
        		              numberOfSubscribers;
        // publication interval
        if(properties.getProperty("publicationInterval") != null) {
        	if(properties.getProperty("publicationInterval").equalsIgnoreCase("CONSTANT")){
        		publicationInterval = CONSTANT;        		
        	} else if (properties.getProperty("publicationInterval").equalsIgnoreCase("EXPONENTIAL")){
        		publicationInterval = EXPONENTIAL;
        	}
        }
        // publication constant
        publicationConstant = properties.getProperty("publicationConstant") != null ?
        		              Double.parseDouble(properties.getProperty("publicationConstant")) :
        		              publicationConstant;
        // reassignment interval
        if(properties.getProperty("reassignmentInterval") != null) {
        	if(properties.getProperty("reassignmentInterval").equalsIgnoreCase("CONSTANT")){
        		reassignmentInterval = CONSTANT;        		
        	} else if (properties.getProperty("reassignmentInterval").equalsIgnoreCase("EXPONENTIAL")){
        		reassignmentInterval = EXPONENTIAL;
        	}
        }
        // reassignment constant
        reassignmentConstant = properties.getProperty("reassignmentConstant") != null ?
        		              Double.parseDouble(properties.getProperty("reassignmentConstant")) :
        		              reassignmentConstant;
        // locality
        locality = properties.getProperty("locality") != null ?
        		   Double.parseDouble(properties.getProperty("locality")) :
        		   locality;
        // locality Power
        localityPower = properties.getProperty("localityPower") != null ?
        		        Double.parseDouble(properties.getProperty("localityPower")) :
        		        localityPower;
        // application seed
        applicationSeed = properties.getProperty("applicationSeed") != null ?
        		          Long.parseLong(properties.getProperty("applicationSeed")) :
        		          applicationSeed;
        // random generator
        rand.setSeed(applicationSeed);		          
	}
	
	/**
	 * Creates an application setting. Its properties are defined by the
	 * class's fields. The application setting is still unassigned to
	 * specific network.
	 * @param sim the current simulation.
	 * @return an application setting.
	 */
	public static Application createApplication(Simulation sim) {

		Job[] jobs; // the jobs
		int n;      // their number
		
		Distribution publicationDistribution;  // publication interval
		Distribution reassignmentDistribution; // reassignment interval

		// create the distribution of the publication interval
		if(publicationInterval == EXPONENTIAL) {
			publicationDistribution = new ExponentialDistribution(rand, publicationConstant);
		}else {
			publicationDistribution = new Constant(publicationConstant);
		}
		
		// create the distribution of the reassignment interval
		if(reassignmentInterval == EXPONENTIAL) {
			reassignmentDistribution = new ExponentialDistribution(rand, reassignmentConstant);
		}else {
			reassignmentDistribution = new Constant(reassignmentConstant);
		}
		
		// create the jobs
		n = numberOfJobs;
		jobs = new Job[n];
		for(int i=0; i<n; i++) {
			jobs[i] = new Job(sim, publicationDistribution);
		}
		
		// create the application object
		return new Application(sim, jobs, reassignmentDistribution, publicationDistribution);
	}	

	/**
	 * Picks a broker from the array respecting the given probabilities.
	 * @param brokers the brokers to choose from. 
	 * @param probs their probabilities.
	 * @return the chosen broker.
	 */
	private static Broker pickBroker(Broker[] brokers, double[] probs){
		
		double sum;	 // the sum of all probability values
		double r; // equal distributed random number between 0 and sum.
		
		// calculate sum
		sum = 0;
		for(int i=0; i< probs.length; i++){
			
			sum += probs[i];
		}
				
		// get next random number
		r = sum * rand.nextDouble();
		
		// find associated broker
		sum = 0;
		for(int i=0; i<probs.length; i++){
			sum += probs[i];
			if(r <= sum){
				// delete its probability to get picked the next time
				probs[i] = 0;
				return brokers[i];
			}
		}
		// neutral value
		return brokers[brokers.length-1];
	}

	/**
	 * Inverts the costs, so that close brokers get higher probabilities.
	 * @param costs the array with communication costs.
	 */
	private void localize(double[] costs, double pow){

		for(int i=0; i < costs.length; i++) {
			if(costs[i]<=0.0d) {
				System.out.println("warning: zero or negative communication costs.");
				costs[i] = 0.0d;
			}else {
				costs[i] = 1/Math.pow(costs[i],pow);
			}
		}
		
//		double min; // minimal costs;
//		double max; // maximal costs;
//		
//		// determine minimal and maximal costs
//		min = Double.MAX_VALUE;
//		max = 0;
//		for(int i=0; i < costs.length; i++){
//			if(costs[i] < min){
//				min = costs[i];
//			}
//			if(costs[i] > max){
//				max = costs[i];
//			}
//		}
//		
//		// mirror values at (min+max)/2
//		for(int i=0; i < costs.length; i++){
//			costs[i] = min + max - costs[i];
//		}
//		
//		// boost the result
//		for(int i=0; i < costs.length; i++){
//			costs[i] = Math.pow(costs[i],pow);
//		}
		
	}
	
	

	/**
	 * Merges arrays of probabilities respecting a locality factor. 
	 * @param factor the locality factor.
	 * @param probs1 probability array.
	 * @param probs2 probability array.
	 * @return a new array with the merged probabilities.
	 */
	private static double[] merge(double factor, double[] probs1, double[] probs2){

		int n;        // arraysize
		double sum1;  // sum of all probabilities of array 1
		double sum2;  // sum of all probabilities of array 2
		double[] rst; // the merged probabilities
		
		// get the size of the smallest array
		n = Math.min(probs1.length,probs2.length);
		
		// determine sums
		sum1 = sum2 = 0;
		for(int i=0; i<n; i++){
			sum1 += probs1[i];
			sum2 += probs2[i];
		}
		
		// ensure sum != 0
		sum1 = sum1 == 0.0d ? 1.0d : sum1;
		sum2 = sum2 == 0.0d ? 1.0d : sum2;
		
		// merge both probabilities and respect their fraction
		rst = new double[n];
		for(int i=0; i<n; i++){
			rst[i]= probs1[i]*factor/sum1 + probs2[i]*(1.0d-factor)/sum2; 
		}
		
		return rst;
	}
	
	/**
	 * Event when a job is reassigned to reflect a changing load situation.
	 */
	private class ReassignmentEvent extends Event {
		
		/**
		 * The job to reassign.
		 */
		Job job;
		
		/**
		 * Creates an event which reassignes this job to other
		 * brokers of the network.
		 * @param job the job to reassign.
		 */
		public ReassignmentEvent(Job job) {
			this.job = job;
		}
		
		/**
		 * Event handler.
		 */
		public void handle() {
			Application.this.assignJob(job, Application.this.net);
		}
	}
	
	/**
	 * The current simulation.
	 */
	private Simulation sim;
	
	/**
	 * The applications jobs.
	 */
	private Job[] jobs;
	
	/**
	 * The distribution of reassignment intervals.
	 */
	private Distribution reassignmentDistribution;
	
	/**
	 * The distribution of publication intervals.
	 */
	//private Distribution publicationDistribution;
	
	/**
	 * The broker's network.
	 */
	private Network net;
	
	/**
	 * Creates a new application object.
	 * @param sim the current simulation.
	 * @param jobs the application's jobs.
	 * @param reassignmentDistribution the distribution of reassignment intervals.
	 * @param publicationDistribution the distibution of publication intervals.
	 */
	private Application(Simulation sim, Job[] jobs, Distribution reassignmentDistribution, Distribution publicationDistribution) {
		this.sim = sim;
		this.jobs = jobs;
		this.reassignmentDistribution = reassignmentDistribution;
		//this.publicationDistribution = publicationDistribution;
	}
	
	
	/**
	 * Assigns the application's job to the brokers of the network.
	 * The jobs are started and also their reassignment is scheduled.
	 * @param net a network of brokers.
	 */
	public void assignTo(Network net) {
		double interval;
		this.net = net;
		for(int i=0; i<jobs.length; i++) {
			assignJob(jobs[i], net);
			interval = reassignmentDistribution.getValue();
			sim.scheduleEventIn(interval, new ReassignmentEvent(jobs[i]));
		}
	}
	
	/**
	 * Assigns a single job to the network. The assigned job is also started.
	 * @param job the job to assign.
	 * @param net the network of brokers.
	 */
	private void assignJob(Job job, Network net){
		
		Broker[] brokers; // the network's brokers
		double[] probs;   // the probabilities to assign an broker
		double[] loads;   // the broker's expected load fraction
		double[] costs;   // the communication costs to an publisher
		
		Broker publisher;     // the job's publisher
		Broker[] subscribers; // the job's subscriber

		int n;       // number of brokers
		int pos;     // publisher position in the broker array
		Iterator it; // to iterate through the brokers
	
		// initialize
		n = net.size();
		brokers = new Broker[n];
		loads = new double[n];
		probs = new double[n];
		costs = new double[n];
		
		// get brokers and their expected load fraction
		it = net.brokerIterator();
		for(int i=0; i<n; i++){ 
			brokers[i] = (Broker)it.next();
			loads[i] = net.getLoad(brokers[i]);
		}
		
		// choose a publisher
		publisher = pickBroker(brokers, loads);
		
		// TODO remove
		//publisher = brokers[67];
		
		
		// get the publishers position and the communication costs to him
		pos = 0;
		for(int i = 0; i<n; i++){
			if(publisher==brokers[i]){
				pos = i;
				loads[i] = 0.0;
				costs[i] = Double.POSITIVE_INFINITY;
			}else {
				loads[i] = net.getLoad(brokers[i]);
				costs[i] = net.getCommunicationCosts(publisher, brokers[i]);
			}
		}

		// choose the subscribers
		subscribers = new Broker[numberOfSubscribers];
		localize(costs, localityPower);
		probs = merge(locality,costs,loads);
		probs[pos]=0;
		for(int i=0; i<subscribers.length; i++){
			subscribers[i] = pickBroker(brokers, probs);
		}
		
		// TODO remove
		System.out.println("Jobs changed");
//		Broker[] bla = {brokers[67],brokers[23], brokers[99], brokers[73], brokers[25], brokers[34], brokers[68]};
//		subscribers = bla;
		// TODO remove
		//System.out.println("Publisher war: "+job.getPublisher()+" Publisher ist: "+publisher);
		
		// set publisher and subscribers and start the job
		job.setPublisher(publisher);
		job.setSubscribers(subscribers);
		job.start();
	}

	// TODO remove
	public Job getJob() {
		return jobs[0];
	}
	
	// TODO remove
	public Job[] getJobs(){
		return jobs;
	}

}