network.java

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

		return overlay.getNumberOfVertices();
	}
	
	public Iterator brokerIterator() {
		return overlay.vertexIterator();
	}
	
    public void send(Destination from, Destination to, Message message){
    	double delay;
    	if(from.isClient() || to.isClient()){
    		delay = 0;
    	}else{
    		// TODO remove
//    		if(message instanceof Notification && !tree.containsEdge((Broker)from,(Broker)to)) {
//    			System.out.println("ERROR: Notification sended from "+from+" to "+to);
//    		}
    		
    		
    		// TODO: insert costs and statistics stuff
    		sim.collectStatisticsSending(from,to,message);
    		
    		delay = getCommunicationDelay((Broker)from,(Broker)to);
    	}
    	sim.scheduleEventIn(delay, new ReceiveEvent(from,to,message));
    }

    public void checkSize() {
    	int size = 0;
    	for(Iterator it = overlay.vertexIterator(); it.hasNext(); ){
    		Broker b =(Broker)it.next();
    		size += b.queueSize();
    	}
    	System.out.println("Elements in Broker queues: "+size);
    	
    	//System.out.println("number: "+number.size()+" interest: "+communicationCosts.size());
    }
    
    public void startHeuristic() {
    	for (Iterator it = brokerIterator(); it.hasNext(); ) {
    		((Broker)it.next()).startHeuristic();
    	}
    }
    
    public void stopHeuristic() {
    	for (Iterator it = brokerIterator(); it.hasNext(); ) {
    		((Broker)it.next()).stopHeuristic();
    	}
    }
    
// IMPORTANT LOAD AN OVERLAY NET    
    
    public static Network loadOverlay(Simulation sim) throws IOException, ParseException{
    	
    	LineNumberReader in;
    	String line;
    	String[] tokens;
    	
    	Network net;
    	Graph physicalNet;
    	Graph overlay;
    	Property load;
    	Property communicationCosts;
    	Property processingCosts;
    	Property linkCosts;
    	Property number;
    	Property distance;
    	Vertex v;
    	Vertex w;
    	Edge e;
    	SimpleHashtable nodes;
    	//Broker broker;
    	
    	int id;
    	int num;
    	int from;
    	int to;
    	
    	double costs;
    	double l;
    	
    	in = new LineNumberReader(new FileReader(netFile));
    	physicalNet = new Graph();
    	overlay = new Graph();
    	number = new Property();
    	processingCosts = new Property();
    	communicationCosts = new Property();
    	processingCosts = new Property();
    	load = new Property();
    	linkCosts = new Property();
    	nodes = new SimpleHashtable();
    	net = new Network(sim,overlay,number,processingCosts, load, communicationCosts);
    	
    	// get beginning
    	while((line = in.readLine())!= null){
    		line = line.trim();
    		if(line.startsWith("#") || line.equals("")){
    			continue;
    		}
    		if(line.startsWith("Nodes")){
    			break;
    		}
    		throw new ParseException("Parse error at line " + 
    					             in.getLineNumber() + 
    					             ": "+ line );   		
     	}
    	
    	// parse nodes
    	while((line = in.readLine()) != null){
    		line = line.trim();
    		if(line.startsWith("#") || line.equals("")){
    			continue;
    		}
    		if(line.startsWith("Edges")){
    			break;
    		}
    		tokens = line.split("\t");
    		try{
    			id = Integer.parseInt(tokens[0]);   			
    			if(tokens[1].equalsIgnoreCase("broker")){
    				num = Integer.parseInt(tokens[2]);
    				if(tokens.length>=4) {
    					costs = Double.parseDouble(tokens[3]);
    				}else {
    					costs = minProcessingCosts;
    				}
    				if(tokens.length>=5) {
    					l = Double.parseDouble(tokens[4]);
    				}else {
    					l = minLoad;
    				}
    				//costs = Double.parseDouble(tokens[3]);
    				//l = Double.parseDouble(tokens[4]);
    				//l = rand.nextDouble()*(maxCommunicationCosts-minCommunicationCosts)+minCommunicationCosts;
    				//l = 5;
    				//costs = maxProcessingCosts-l;
    				v = new Broker(sim, net);
    				v.setProperty(number, new Integer(num));
    				v.setProperty(processingCosts, new Value(costs));
    				v.setProperty(load,new Value(l));
    				overlay.addVertex(v);
    			}else{
    				v = new Vertex();
    			}
    			physicalNet.addVertex(v);
    			nodes.put(new Integer(id),v);
    		}catch(Exception x){
    			throw new ParseException("Parse error at line " + 
    					                 in.getLineNumber() + 
    					                 ": "+x);
    		}
    	}
    	
    	// parse physical edges
    	while((line = in.readLine()) != null){
    		line = line.trim();
    		if(line.startsWith("#") || line.equals("")){
    			continue;
    		}
    		if(line.startsWith("Overlay")){
    			break;
    		}
    		tokens = line.split("\t");
    		try{
    			from = Integer.parseInt(tokens[0]);
    			to = Integer.parseInt(tokens[1]);
    			costs = Double.parseDouble(tokens[2]);
    			v = (Vertex)nodes.get(new Integer(from));
    			w = (Vertex)nodes.get(new Integer(to));
    			e = new Edge(v,w);
    			e.setProperty(linkCosts, new Double(costs));
    			physicalNet.addEdge(e);
     		}catch(Exception x){
    			throw new ParseException("Parse error at line " + 
		                 in.getLineNumber() + 
		                 ": "+x);
     		}
    	}
    	
    	// parse overlay edges
    	while((line = in.readLine()) != null){
    		line = line.trim();
    		if(line.startsWith("#") || line.equals("")){
    			continue;
    		}
    		if(line.startsWith("Overlay")){
    			break;
    		}
    		tokens = line.split("\t");
    		try{
    			from = Integer.parseInt(tokens[0]);
    			to = Integer.parseInt(tokens[1]);
    			costs = Double.parseDouble(tokens[3]);
    			v = (Broker)nodes.get(new Integer(from));
    			w = (Broker)nodes.get(new Integer(to));
    			e = new Edge(v,w);
    			e.setProperty(communicationCosts, new Double(costs));
    			overlay.addEdge(e);
     		}catch(Exception x){
    			throw new ParseException("Parse error at line " + 
		                 in.getLineNumber() + 
		                 ": "+x);
     		}
    	}
    	
    	// close the file
    	in.close();
    	
    	// calculate communication costs of missing edges
    	distance = new Property();
    	int x=0;
    	for(Iterator it = overlay.vertexIterator(); it.hasNext(); ){
    		v = (Vertex)it.next();
    		distance.clear();
    		//System.out.println("Dijkstra: "+x++);
    		dijkstra(physicalNet, v, linkCosts, distance);
    		for(Iterator jt = overlay.vertexIterator(); jt.hasNext(); ){
    			w = (Vertex)jt.next();
    			if(v==w || overlay.containsEdge(v,w)){
    				continue;
    			}
    			e = new Edge(v,w);
    			e.setProperty(communicationCosts, new Value(((Double)w.getProperty(distance)).doubleValue()));
    			//System.out.print(((Double)w.getProperty(distance)).doubleValue()+" ");
    			//System.out.println();
    			overlay.addEdge(e);
    		}
    	}
    	distance.clear();
    	return net;
    }

     private static void dijkstra(Graph g, Vertex s, Property linkCosts, Property distance){
    	LinkedList queue;
    	Vertex v;
    	Vertex node;
    	Edge e;
    	double min;
    	double dist;
    	double cost;
    	
    	
    	
    	queue = new LinkedList();
    	queue.add(s);

    	s.setProperty(distance,new Double(0));
    	
    	while(!queue.isEmpty()){
    		// get nearest marked noce
    		node = null;
        	min = Double.MAX_VALUE;
    		for(Iterator it = queue.iterator(); it.hasNext(); ){
    			v = (Vertex)it.next();
    			dist = ((Double)v.getProperty(distance)).doubleValue();
    			if(dist < min){
    				min = dist;
    				node = v;
    			}
    		}
    		
    		queue.remove(node);
    		
    		// relax
    		for(AdjacencyIterator ait = g.adjacencyIterator(node); ait.hasNext(); ){
    			ait.next();
    			v = ait.vertex();
    			e = ait.edge();
    			cost = ((Double)e.getProperty(linkCosts)).doubleValue();
    				
    			if(v.hasProperty(distance)){
    				dist = ((Double)v.getProperty(distance)).doubleValue();
    				if(min+cost < dist){
    					v.setProperty(distance,new Double(min+cost));
    				}
    			}else{
    				v.setProperty(distance, new Double(min+cost));
    				queue.add(v);
    			}
    		}
     	}
    }
    
    public void change(Network net) {
    	
    	Broker bOld;
    	Broker bOld2;
    	Broker bNew;
    	Edge eOld;
    	Edge eNew;
    	
    	SimpleHashtable brokers;
    	Integer num;	
    	Value v;
    	double p;
    	double l;
    	double c;
    	
    	// TODO remove
    	System.out.println("Network ist changed");
    	
    	if(changeEvent != null) {
    		throw new IllegalStateException("concurrent modification");
    	}
    	
    	brokers = new SimpleHashtable();
    	for (Iterator it = this.brokerIterator(); it.hasNext(); ) {
    		bOld = (Broker)it.next();
    		num = (Integer)bOld.getProperty(number);
    		brokers.put(num,bOld);
    	}
    	
    	for (Iterator it = net.brokerIterator(); it.hasNext(); ) {
    		bNew = (Broker)it.next();
    		p = net.getProcessingCosts(bNew);
    		l = net.getLoad(bNew);
    		bOld = (Broker)brokers.get(new Integer(net.getNumber(bNew)));
    		if (bOld != null) {
    			v = (Value)bOld.getProperty(load);
    			v.from = v.current;
    			v.to = l;
    		
    			v = (Value)bOld.getProperty(processingCosts);
    			v.from = v.current;
    			v.to = p;
    		}
    	}
    	
    	for (Iterator it = net.overlay.edgeIterator(); it.hasNext(); ) {
    		eNew = (Edge)it.next();
    		c = net.getCommunicationCosts((Broker)eNew.s, (Broker)eNew.t);
    		bOld = (Broker)brokers.get(new Integer(net.getNumber((Broker)eNew.s)));
    		bOld2 = (Broker)brokers.get(new Integer(net.getNumber((Broker)eNew.t)));
    		eOld = bOld != null && bOld2 != null ?
    			   this.overlay.getEdge(bOld, bOld2) :
    			   null;
            if (eOld != null) {
            	v = (Value)eOld.getProperty(communicationCosts);
            	v.from = v.current;
            	v.to = c;
            }
    	}
    	
    	changeEvent = new ChangeEvent(changeSteps,changeInterval);
    	sim.scheduleEventNow(changeEvent);
    }

    private class ChangeEvent extends Event{
    	int i = 1;
    	int steps;
    	double interval;
    	
    	public ChangeEvent(int steps, double interval) {
    		this.steps = steps;
    		this.interval = interval;
    	}
    	
    	public void handle() {
    		Network.this.change(i,steps);   		
    		if(i < steps) {
    			Network.this.sim.scheduleEventIn(interval,this);
    		}else{
    			Network.this.changeEvent = null;
    		}
    		i++;
    	}
    }
    
    
    private void change(int step, int max) {
    	Broker b;
    	Edge e;
    	Value v;
    	
    	for (Iterator it = this.brokerIterator(); it.hasNext(); ) {
    		b = (Broker)it.next();
    		v = (Value)b.getProperty(load);
    		v.current = v.from + ((double)step/(double)max)*(v.to - v.from);
    		v = (Value)b.getProperty(processingCosts);
    		v.current = v.from + ((double)step/(double)max)*(v.to - v.from);
    	}
    	
       	for (Iterator it = overlay.edgeIterator(); it.hasNext(); ) {
    		e = (Edge)it.next();
    		v = (Value)e.getProperty(communicationCosts);
    		v.current = v.from + ((double)step/(double)max)*(v.to - v.from);
    	}
       	
       	// TODO remove
       	if(step == max){
       		System.out.println(sim.getTime()+": Network is fully changed");
       	}
    }
    
    
    
    // TODO Zum Testen.
    Property ints = new Property();
    public void createInterestTree(Application app){
    	
    	Property component = new Property();
    	
    	Job[] jobs = app.getJobs();
    	
    	for(int i=0; i<jobs.length; i++) {
    		Broker[] brokers = new Broker[10];
    		brokers[0] = jobs[i].getPublisher();
    		Broker[] subscribers = jobs[i].getSubscribers();
    		for(int j=0; j<subscribers.length; j++){