environmentsimulatorimpl.java

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

		{
/*			//Get downstream handle
			String strNextConnection = (String)XMLConfigurator.getParameter("stages|stage" + nstage +"|connection" + nplacement);
			String [] places = strNextConnection.split(":");
			int nextStage = Integer.parseInt(places[0].substring(5)); 
			int nextPlacement = Integer.parseInt(places[1].substring(9)); 

			URL urlHandle = (URL)XMLConfigurator.getParameter("stages|stage" + nextStage + "|placement" + nextPlacement);
			if(urlHandle == null)
			{
				log.error("can't find the downstream service of stage " + nstage + " and placment " + nplacement);
				return null;
			}
			newDownstreamHandle[0] = urlHandle.toString(); */
			return strConfig;
		}
		else
		{
			log.fatal("strConfig has a bad format");
			return null;
		}
		
	}
	
	private synchronized String generatingConfigString(GraphAlgorithm ga_final)
	{
		Vector v;
		int curStage, nLevel, tempJ, fakeNumStages, nSources, i;
		nSources = ((Integer)XMLConfigurator.getParameter("stages|stage1|numPlacements")).intValue();
		
		String [] strSourceIPs;
		String strDest, strTemp;
		strSourceIPs = new String[nSources + 1];
		int nDepth[] = new int[nSources + 1];
		DefaultMutableTreeNode [] trees = new DefaultMutableTreeNode[nSources];
		DefaultMutableTreeNode treenode, treeRoot, treenode_final;

		//Get the destination
		strTemp = ((URL)XMLConfigurator.getParameter("stages|stage" + numStages + "|placement1")).toString();
		strDest = Utilities.getIPAddress(strTemp);

		//Initialize the DynamicEnv's numVisits
		for(i = 0; i < ga_final.n; i ++)
			ga_final.d[i].numVisits = 0;

		int [] numPlacementsInStage = new int[100];
		for(i = 1; i < 100; i ++)
			numPlacementsInStage[i] = 0;

		Integer tempInt;
		fakeNumStages = -1;
		log.debug("come to here");
		for(int out_loop = 1; out_loop <= nSources; out_loop ++)
		{
			strTemp = ((URL)XMLConfigurator.getParameter("stages|stage1|placement"+out_loop )).toString();
			strSourceIPs[out_loop] = Utilities.getIPAddress(strTemp);
			log.debug(strSourceIPs[out_loop]);
			if((treenode = ga_final.getTreeNode(strSourceIPs[out_loop], 1, out_loop)) == null)
			{
				if((treenode =  ga_final.getTreeNode(strSourceIPs[out_loop], -1, -1)) == null)
				{
					log.error("can't find node");
					continue;
				}
			}
			//As regard with the stage 1st, we don't need to specify anything
			curStage = 1;
			nDepth[out_loop] = treenode.getLevel();
			if(nDepth[out_loop] + 1> fakeNumStages)
				fakeNumStages = nDepth[out_loop] + 1;

			int myLevel, myDepth, myStage1, myStage2;
			myLevel = myDepth = myStage1 = myStage2 = -1;
			do{
				myLevel = treenode.getLevel();
				//Determine what stages in this tree node
				myDepth = nDepth[out_loop] + 1 - myLevel;
				myStage1 = myStage2 = -1;
				if(nDepth[out_loop] + 1 >= numStages)
				{
				//there is only one stage in this tree node
					if(nDepth[out_loop] + 1 == numStages)
						myStage1 = nDepth[out_loop] + 1 - myLevel;
					else
					{
						if(myDepth == nDepth[out_loop] + 1)
						//This is the last stage
							myStage1 = numStages;
						else if( myDepth < numStages)
							myStage1 = myDepth;
						else
							myStage1 = myDepth + 1;
					}
					myStage2 = myStage1;
				}
				else
				{
					if(myDepth == 2)
					{
						myStage1 = 2;
						myStage2 = numStages - (nDepth[out_loop] + 1) + myStage1;
					}
					else if(myDepth == 1)
						myStage1 = myStage2 = 1;
					else
						myStage1 = myStage2 = myDepth + (numStages - (nDepth[out_loop] + 1));
				}
				log.debug(((DynamicEnv)(treenode.getUserObject())).name);
				log.debug("mystage1:"+myStage1+" mystage2:"+myStage2);
				log.debug("myLevel:"+myLevel+" nDepth:"+nDepth[out_loop]);

				DynamicEnv dNode = (DynamicEnv)(treenode.getUserObject());
				int curVisit = dNode.numVisits;
				log.debug("curVisit:"+curVisit);
				dNode.numVisits ++;
				dNode.visitArray[curVisit] = new Visit(treenode, curVisit, myStage1, myStage2);

				//not necessary
				// |
				// |
				// +
				// The match function in fact change visitArray[]
				for(int lpIndex = 0; lpIndex <= curVisit - 1; lpIndex ++)
					if(dNode.visitArray[curVisit].Match(dNode.visitArray[lpIndex]))
						break;
				// ^
				// |
				// |
				//not necessary



/*				if(myStage1 == myStage2 && myStage1 == 1) //the first stage
				{
					 dNode.visitArray[curVisit].strPlacements[0] = ((URL)XMLConfigurator.getParameter("stages|stage1|placement"+out_loop )).toString();
				}
*/
				treenode = (DefaultMutableTreeNode)treenode.getParent();
			}while(treenode != null);
		}

		//Query the individual information center to get the IP address of available nodes
		log.debug("come to WriteXML");

		WriteXml xmlWriter;
		try{ 
			xmlWriter = new WriteXml();
			xmlWriter.initXMLConfig_for_dynamic();
		}
		catch(Exception e)
		{
			log.error(e);
			return null;
		}

		//Initialize the DynamicEnv's numVisits
		for(i = 0; i < ga_final.n; i ++)
			ga_final.d[i].numVisits = 0;
		for(i = 1; i <= nSources; i ++)
		{
			if((treenode = ga_final.getTreeNode(strSourceIPs[i], 1, i)) == null)
			{
				if((treenode =  ga_final.getTreeNode(strSourceIPs[i], -1, -1)) == null)
				{
					log.error("can't find node");
					continue;
				}
			}
/*			treenode.getLevel() + 1 > nStages)
				bSpecialBranch= true;
			else
				bSpecialBranch= false; */
			WriteToConfigFile(xmlWriter, treenode, treenode.getLevel(), numPlacementsInStage); //this is a recursive function
		}
		for(i = 2; i <= fakeNumStages; i ++)
		{
			if(i == numStages)
				continue;

			tempInt = new Integer(numPlacementsInStage[i]);
			xmlWriter.addPara(i, "numPlacements", "Integer", tempInt.toString());
		}
		tempInt = new Integer(fakeNumStages);
		xmlWriter.addPara("fakeNumStages", "Integer", tempInt.toString());
		//Write the special_class information
		String specialclass_name, specialclass_location;
		specialclass_name = (String)XMLConfigurator.getParameter("special_class|class_name");
		specialclass_location = ((URL)XMLConfigurator.getParameter("special_class|location")).toString();
		xmlWriter.addSpecialClass(specialclass_name, specialclass_location);
		//Write about Envirnoment service
		String envHandle = ((URL)XMLConfigurator.getParameter("EnvironmentSimulator")).toString();
		xmlWriter.addObject("EnvironmentSimulator", "java.net.URL", "String", envHandle);

		//Debug
		log.debug("finish WriteXML");
//		XMLUtils.PrettyElementToStream(xmlWriter.getApp(), System.out);
//		XMLUtils.ElementToStream(xmlWriter.getApp(), System.out);
		//Launch the stages
		//********************
		//********************
		//Write the configuration file to StreamClient
		//Create a PipedInputStream and a PipedOutputStream
		/*PipedOutputStream outstream = new PipedOutputStream();
		PipedInputStream instream = new PipedInputStream(outstream);
		XMLUtils.ElementToStream(xmlWriter.getApp(), outstream);
		outstream.connect(instream);*/
		log.debug("generating a configuration string ");
		String strCnfResults;
/*			String strCnfResults = XMLUtils.ElementToString(xmlWriter.getApp());
                        byte [] tempBytes = strCnfResults.getBytes();
			ByteArrayInputStream byteIn = new ByteArrayInputStream(tempBytes);*/

/*			StringWriter writer = new StringWriter();
			XMLUtils.PrettyElementToWriter(xmlWriter.getApp(), writer);

			strCnfResults = writer.toString();*/
		strCnfResults = XMLUtils.ElementToString(xmlWriter.getApp());
		log.debug(strCnfResults);
		//polish the strCnfResults
		int index;
		String result = new String("");
		String result1;
				
		while(true)
		{
			index = strCnfResults.indexOf('>');
			if(index < 0)
				break;
			result1 = strCnfResults.substring(0, index + 1);
			if(result1.lastIndexOf("</") < 0)
				result += result1;
			else
				result += result1 + "\n";

			if(index + 1 == strCnfResults.length())
				break;
			strCnfResults = strCnfResults.substring(index + 1, strCnfResults.length());
		}
		log.info(result);
		return result;
		/*log.debug(result);
	        byte [] tempBytes = result.getBytes();
		for(int f = 0; f < tempBytes.length; f ++)
		System.out.println((char)tempBytes[f]);
       	        ByteArrayInputStream byteIn = new ByteArrayInputStream(tempBytes);
		l.start(byteIn);*/
	}

	private String WriteToConfigFile(WriteXml xmlWriter, DefaultMutableTreeNode treenode, int nDepth, int [] numPlacementsInStage)
	{
		if(treenode == null)
			return null;
		log.debug(((DynamicEnv)(treenode.getUserObject())).name);

		DynamicEnv dNode = (DynamicEnv)(treenode.getUserObject());
		dNode.remove(-1,-1);
		int []newPlacement;
		int curVisit = dNode.numVisits;
		log.debug("curVisit:"+curVisit);
		dNode.numVisits ++;
		Visit vis = dNode.visitArray[curVisit];

		Integer tempInt;
		boolean bSpecialStage = false;
		tempInt = (Integer)XMLConfigurator.getParameter("numStages");
		int numStages = tempInt.intValue();
		int nstage, index ;

		newPlacement = new int[vis.myStage2 - vis.myStage1 + 1];

		for(nstage = vis.myStage1; nstage <= vis.myStage2; nstage ++)
		{
			index = nstage - vis.myStage1;
			log.debug("my visit no. is" + vis.visitnodeIndex[index]);
			log.debug("actual visit no. is" + vis.no);
			log.debug("myStage1:" + vis.myStage1 + "["+nstage+"] myStage2:" + vis.myStage2); 
			if(nstage == 1)
			{ //Placement info. has been done, go to the next part of this function
				numPlacementsInStage[nstage] ++;
				newPlacement[index] = assignPlacement(dNode, treenode, 1);
				continue;
			}
			else if(nstage == numStages)
			{//Everything is known and done, return;
				numPlacementsInStage[nstage] = 1;
				newPlacement[index] = 1;
//				vis.strPlacements[index] = "stage"+nstage+":placement1";
				log.debug("the last stage, go on to the next step");
				break;
				//log.debug("returning: stage"+ nstage + ":placement1");
				//return "stage"+nstage+":placement1";
			}

			//log.debug("my visit no. is" + vis.visitnodeIndex[index]);
			//log.debug("actual visit no. is" + vis.no);
			if(vis.visitnodeIndex[index] < vis.no)
			{//the placement and connection has been done by previous visit
			 //return the previous connection information.
			 	int visitIndex = vis.visitnodeIndex[index];
				DefaultMutableTreeNode tempNode = (DefaultMutableTreeNode)treenode.getParent();
				while(tempNode != null)
				{
					 ((DynamicEnv)(tempNode.getUserObject())).numVisits++;
					 tempNode = (DefaultMutableTreeNode)tempNode.getParent();
				}

				//get the placement information from dNode
				InstanceIndex ii = dNode.getInstanceIndex(treenode, nstage);
				if(ii == null)
				{
					log.error("some thing error with the structure");
					return null;
				}

				log.debug("returning: stage"+ nstage);
				log.debug("returning: should be " + nstage + " now: " + ii.nstage+ ":placement"+ii.nplacement);
				return "stage"+ii.nstage+":placement"+ii.nplacement; //return gaNode.visitArray[visitIndex].strConnections[index];
			}


			vis.strPlacements[index] = dNode.name;
			newPlacement[index] = assignPlacement(dNode, treenode, nstage);
			numPlacementsInStage[nstage] ++;
			log.debug(vis.strPlacements[index]);
			log.debug("the assigned placement is:" + newPlacement[index]);
			
			//write to xml configuration file
			xmlWriter.addObject(nstage, "placement"+newPlacement[index],
					   "java.net.URL",
					   "String",
					   "http://"+  vis.strPlacements[index] +":8080/ogsa/services/Stream/StreamService"
					   );
			//Add a "index_service" to the configuration file
/*			xmlWriter.addPara(nstage, "index_service"+stagePlacements[nstage], 
					   "String",
					   "http://"+  vis.strPlacements[index] +":8080/ogsa/services/Stream/StreamService"
					   ); */

		}


		String connection_name, myConnectionValue, class_name, location;
		int nStoreFirstStagePlacement = 0, nextStage = 0;

		//What is the first stage in this branch
		//Get the connection information.
		int new_bandwidth = 0;
		double new_util = 0.0;
		for(nstage = vis.myStage1; nstage <= vis.myStage2; nstage ++)
		{
			log.debug("myStage1:" + vis.myStage1 + "["+nstage+"] myStage2:" + vis.myStage2); 
			index = nstage - vis.myStage1;

			connection_name = "connection" + newPlacement[index];
			//get the placement information of the next stage
			if(nstage == vis.myStage2) //The last stage in this node
			{
				String myName = ((DynamicEnv)(treenode.getUserObject())).name;
				DefaultMutableTreeNode parent = (DefaultMutableTreeNode)treenode.getParent();
				if(parent == null) //the last stage
					break;
				String myParentName = ((DynamicEnv)(parent.getUserObject())).name;
				if(myParentName.equals(myName))
				{
					new_bandwidth = -1;
					new_util = 0.0;
				}
				else if(ga_init.findEdge(myName, myParentName) < 0)
				{
					log.error("can't find " + myName + " and " + myParentName + " 's edge");
					continue;
				}
				new_bandwidth = ga_init.getEdgeLength(myName, myParentName);
				new_util = ga_init.getEdgeUtil(myName, myParentName);

				//Add a "bandwidth" to the configuration file
				xmlWriter.addPara(nstage, "bandwidth"+newPlacement[index], "String", Integer.toString(new_bandwidth)+":" + Double.toString(new_util));

				//We need to recurviely call WriteToConfigFile
				//and pass the parent node to the function
				myConnectionValue = WriteToConfigFile(xmlWriter, parent, nDepth, numPlacementsInStage);
				if(myConnectionValue == null)
				{
					vis.strConnections[index] = myConnectionValue;
					return "stage" + nstage + ":placement1";
				}

			}
			else//Get the the information from the node itself
			{
				myConnectionValue = "stage"+(nstage+1)+":placement"+newPlacement[index+1];
				//Add a "bandwidth" to the configuration file
				log.debug("adding bandwidth"+newPlacement[index]+ ":-1 to stage" + nstage);
				xmlWriter.addPara(nstage, "bandwidth"+newPlacement[index], "String", "-1");
			}

			//Get the classname and location
			vis.strConnections[index] = myConnectionValue;
			if(nstage > numStages)
			{
				int oldFakeNumStages = ((Integer)XMLConfigurator.getParameter("fakeNumStages")).intValue();
				String specialclass_name, specialclass_location;
				if(oldFakeNumStages > numStages)
				{
					specialclass_name = (String)XMLConfigurator.getParameter("stages|stage" + oldFakeNumStages + "|class|class_name");
					specialclass_location = ((URL)XMLConfigurator.getParameter("stages|stage" + oldFakeNumStages + "|class|location")).toString();
				}
				else
				{
					specialclass_name = (String)XMLConfigurator.getParameter("special_class|class_name");
					specialclass_location = ((URL)XMLConfigurator.getParameter("special_class|location")).toString();
				}
				xmlWriter.addClass(nstage, specialclass_name, specialclass_location);
			}

			xmlWriter.addPara(nstage, connection_name, "String", myConnectionValue);
			log.debug(connection_name+ " "+myConnectionValue);
		}
		((DynamicEnv)(treenode.getUserObject())).bVisited = true;
		return "stage"+vis.myStage1+":placement"+newPlacement[0];
	}

/*	private int addNewPlacement(Hashtable [] stagePlacements, String name, int nstage)
	{