simulatorengine.java

本人历尽千辛万苦找的clustream中的jar包

package org.osu.ogsa.stream.services;
/*import java.util.Random;
import java.lang.Math;
import java.lang.System;
import org.apache.log4j.*; */

import org.osu.ogsa.stream.util.*;	
import org.osu.ogsa.stream.util.xmlconfig.*;
import org.osu.ogsa.stream.services.EnvironmentSimulatorImpl;
import org.osu.ogsa.stream.StreamMonitor.StreamMonitorPortType;

import java.io.*;
import java.nio.*;
import java.nio.channels.*;
import java.nio.channels.spi.*;
import java.nio.charset.*;
import java.net.*;
import java.util.*;
import java.lang.*;
import org.apache.log4j.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.axis.client.Stub;
import javax.swing.tree.DefaultMutableTreeNode;
import org.apache.axis.utils.XMLUtils;
import org.w3c.dom.Document;
import org.w3c.dom.DOMException;
import org.w3c.dom.Element;
import org.w3c.dom.Node;

public class SimulatorEngine implements Runnable //extends Thread
{
        //private static Category log = Category.getInstance(SimulatorEngine.class.getName());
        static Log log = LogFactory.getLog(SimulatorEngine.class.getName());
	/*
	 * orderStage[1] = 2 means that system has assigned the 2nd place for the stage 1
	 */  
//	private static int iPort = 58888;//, 57777, 56666, 55555;
	
	private int numEdges, numNodes;
	private int nEnvChangingTimes = DefConstants.BANDWIDTH_VARING_TIMES;
	private int sleeping_time = 1000; //1 millisecond
	private EnvironmentSimulatorImpl simulatorSrv;
	private static double [] nUtils = {0.3, 0.2, 0.45, 0.7};
	
	SimulatorEngine(EnvironmentSimulatorImpl simulatorSrv)
	{
		this.simulatorSrv = simulatorSrv;
		numEdges = simulatorSrv.ga_init.m;
		numNodes = simulatorSrv.ga_init.n;
	}

	public void run()
	{
		int i, j, nUtilIndex;

		Date forRand = new Date();
		Random rand = new Random(forRand.getTime());
		Random [] randArray = new Random[numEdges];

		for(i = 0; i < numEdges; i++)
			randArray[i] = new Random(rand.nextLong());

		Random rand_pickup_util = new Random(forRand.getTime() + 10);
		Random rand_generate_util = new Random(forRand.getTime() + 100);
		double dNetworkUtil;
		StreamMonitorPortType [] monitors;


		double probability = (double)nEnvChangingTimes/((double)DefConstants.LIFETIME*1000/(double)sleeping_time);
	//	double probability = 0.01;//(double)nEnvChangingTimes/((double)DefConstants.LIFETIME*1000/(double)(sleeping_time*numEdges));
		log.debug("the probability to change the network bandwidth is " + probability);

		int rndd_plus, rndd_minus, opp_i;
		rndd_plus = rndd_minus = opp_i= 0;
		
		try{
			while(true)
			{
				Thread.sleep(sleeping_time);
				//check if all monitor services signed in
				if(!simulatorSrv.isAllMonitorsSignedIn())
					continue;
				//start to change the bandwidthes and cpu loades
				for(i = 0; i < numEdges ; i ++)
				{
					if(Utilities.High_Precsion_Biased_Coin(randArray[i], probability) == DefConstants.HEAD)
					{
						//changing the bandwidth
						nUtilIndex = Utilities.randnum(rand_pickup_util, 5) - 1;

						if(nUtilIndex == 4)
							//random generate a util
							dNetworkUtil = rand_generate_util.nextDouble();
						else
							dNetworkUtil = nUtils[nUtilIndex];
						Double dNew = new Double(dNetworkUtil);
						Double dOld = new Double(simulatorSrv.ga_init.getUtil(i));
					//	log.debug("the old util:" + dOld + " the new util:" + dNew);
						if(!dNew.equals(dOld))
						{
							//write the util to the edge and modify the fake network bandwidth
							simulatorSrv.ga_init.setUtil(i, dNetworkUtil);
							//modified the opposite direction
							rndd_plus = simulatorSrv.ga_init.e[i].rndd_plus;
							rndd_minus = simulatorSrv.ga_init.e[i].rndd_minus;
							if(rndd_plus < 0 || rndd_minus < 0)
							{
								log.error("wrong information");
								continue;
							}

							opp_i = simulatorSrv.ga_init.findEdge(simulatorSrv.ga_init.v[rndd_minus].name, simulatorSrv.ga_init.v[rndd_plus].name);
							log.debug("i:" + i + "v.s." + opp_i);
							if(opp_i >= 0)
								simulatorSrv.ga_init.setUtil(opp_i, dNetworkUtil);

							//Notify the monitor service to simulate the new bandwidth
							monitors = simulatorSrv.getMonitor(i);
							if(monitors == null)
							{
//								log.info("can't get the monitor's handle, edge: " + i);
								continue;
							}
							//Get the downstream's handle
							for(j = 0; monitors[j] != null; j ++)
							{
							String strDownstreamHandle = (String)simulatorSrv.hashStrDownstream.get((Object)monitors[j]);
							if(strDownstreamHandle == null)
							{
								log.info("can't get the downstream's handle:" + strDownstreamHandle);
								continue;
							}
	                                                String downName = Utilities.getIPAddress(strDownstreamHandle);
							if(downName == null)
							{
								log.error(strDownstreamHandle + "'s ip address is null, that is impossible");
								continue;
							}
							if(!downName.equals(simulatorSrv.ga_init.v[rndd_minus].name))
								continue;
							log.debug("change the bandwidth of the service of which downstream is " + strDownstreamHandle);
							log.debug("the old util:" + dOld + " the new util:" + dNew);
							monitors[j].netBandwidthNotification(simulatorSrv.ga_init.e[i].len, dNetworkUtil, strDownstreamHandle);
							break;
							}
						}
					}
	/*				else{
						log.debug("no possible");
					} */
				}
	/*			for(i = 0; i < numNodes; i ++)
				{
					//modify cpu load
				} */
			}
		}
		catch(Exception e)
		{
			log.error(e);
		}
	}
}