resourcemng.java

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

			outstream.connect(instream);*/
			if(DefConstants.IFLAUNCHING)
			{
				log.debug("launching the stages");
				Launcher l = new Launcher();
				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);
				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.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);
			}
		}
		catch(Exception e)
		{
			System.out.println(e);
		}
	}
	

	public static String WriteToConfigFile(WriteXml xmlWriter, DefaultMutableTreeNode treenode, int nDepth)
	{
		if(treenode == null)
			return null;
		log.debug(((GANode)(treenode.getUserObject())).name);

		GANode gaNode = (GANode)(treenode.getUserObject());
		int curVisit = gaNode.numVisits;
		gaNode.numVisits ++;
		Visit vis = gaNode.visitArray[curVisit];

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

		for(nstage = vis.myStage1; nstage <= vis.myStage2; nstage ++)
		{
			index = nstage - vis.myStage1;
			if(nstage == 1)
			{ //Placement info. has been done, go to the next part of this function
				orderStage[nstage] ++;
				continue;
			}
			else if(nstage == numStages)
			{//Everything is known and done, return;
				vis.strPlacements[index] = "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)
				{
					 ((GANode)(tempNode.getUserObject())).numVisits++;
					 tempNode = (DefaultMutableTreeNode)tempNode.getParent();
				}

				return "stage"+nstage+":placement"+orderStage[nstage]; //return gaNode.visitArray[visitIndex].strConnections[index];
			}

			int nTypeRes;
			if(numStages < nstage)
				nTypeRes = 2; //Memory are more desirable
			else
			{
				tempInt =  (Integer)XMLConfigurator.getParameter("stages|stage" + nstage + "|resources_needed");

				log.debug("stages|stage" + nstage + "|resources_needed");
				log.debug(tempInt);
				nTypeRes = tempInt.intValue();
			}

			String strXPathExp;
			switch(nTypeRes)
			{
				default:
				case 0: //stands for "no idea"
					strXPathExp = "//ce:Host[//ce:Host/ce:ProcessorLoad[@ce:Last1Min<"+DefConstants.GENERAL_LAST15MIN+" and @ce:Last1Min>=0]][//ce:Host/ce:MainMemory[@ce:RAMAvailable>="+DefConstants.GENERAL_RAMAVAILABLE+"]][//ce:Host/ce:Processor[@ce:ClockSpeed>"+DefConstants.GENERAL_CLOCKSPEED+"]]";
					break;
				case 1://1 stands for "computation intensive"
					strXPathExp = "//ce:Host[//ce:Host/ce:ProcessorLoad[@ce:Last1Min<"+DefConstants.SPECIAL_LAST15MIN+" and @ce:Last1Min>=0]][//ce:Host/ce:MainMemory[@ce:RAMAvailable>=("+(DefConstants.GENERAL_RAMAVAILABLE - 20)+")]][//ce:Host/ce:Processor[@ce:ClockSpeed>"+DefConstants.SPECIAL_CLOCKSPEED+"]]";
					break;
				case 2:// 2 stands for "memory desired"
					strXPathExp = "//ce:Host[//ce:Host/ce:ProcessorLoad[@ce:Last1Min<"+DefConstants.GENERAL_LAST15MIN+" and @ce:Last1Min>=0]][//ce:Host/ce:MainMemory[@ce:RAMAvailable>="+DefConstants.SPECIAL_RAMAVAILABLE+"]][//ce:Host/ce:Processor[@ce:ClockSpeed>"+DefConstants.GENERAL_CLOCKSPEED+"]]";
					break;
				case 3://3 stands for "both cpu and memory are intensive"
					strXPathExp = "//ce:Host[//ce:Host/ce:ProcessorLoad[@ce:Last1Min<"+DefConstants.SPECIAL_LAST15MIN+" and @ce:Last1Min>=0]][//ce:Host/ce:MainMemory[@ce:RAMAvailable>="+DefConstants.SPECIAL_RAMAVAILABLE+"]][//ce:Host/ce:Processor[@ce:ClockSpeed>"+DefConstants.SPECIAL_CLOCKSPEED+"]]";
					break;
			}
			String service_handle = "http://" + ((GANode)(treenode.getUserObject())).name+":8080/ogsa/services/base/index/IndexService";
			log.debug(strXPathExp);
			log.debug(service_handle);
			int intTry = 3;
			while(intTry > 0)
			{
			Element [] elementRes = QueryServiceDataByXPath.Query("MyHostScript", service_handle, strXPathExp);
			if(elementRes == null)
			{
				log.error("there is no node matching the resource requirements");
				log.error("try it again....");
				intTry --;
				continue;
				//Further processing.....????????
//				 ((GANode)(treenode.getUserObject())).strPlacement[elementIndex] = ((GANode)(treenode.getUserObject())).name;
//				 
//				 ((GANode)(treenode.getUserObject())).table[elementIndex] = orderStage[nstage];
			}
			else
			{
				 //((GANode)(treenode.getUserObject())).strPlacement[elementIndex] = elementRes[0].getAttribute("ce:Name");
//			/	 log.debug(((GANode)(treenode.getUserObject())).strPlacement[elementIndex]);
				int length = elementRes.length;
				Date forRand = new Date();
				Random rand = new Random(forRand.getTime());
				
				int rand_int = Math.abs(rand.nextInt()) % length;
				log.debug("the " + rand_int+"node is selected");
			
				 vis.strPlacements[index] = elementRes[rand_int].getAttribute("ce:Name");
				orderStage[nstage] ++;
				 log.debug(vis.strPlacements[index]);
				 break;
//				 ((GANode)(treenode.getUserObject())).table[elementIndex] = orderStage[nstage];
				//Lots of things needed to be handled:
				//1. How to avoid the occurance of assigning some stages to the same machine?
				//2. How to effeciently query the information server. I.e. we can query the information server one time when we find two stages need the same resources.
				//....
			}
			}
			if(intTry ==0)
			{
				 vis.strPlacements[index] = "there is no node matching the resource requirement";
				 orderStage[nstage] ++;
			}
			//write to xml configuration file
			xmlWriter.addObject(nstage, "placement"+orderStage[nstage], 
					   "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"+orderStage[nstage], 
					   "String", service_handle);
//			XMLUtils.PrettyElementToStream(xmlWriter.getApp(), System.out);

		}


		String connection_name, myConnectionValue, class_name, location;
		int nStoreFirstStagePlacement = 0, nextStage = 0;
//		GANode tempGANode = (GANode)(treenode.getUserObject());

		//What is the first stage in this branch
		//Get the connection information.
		for(nstage = vis.myStage1; nstage <= vis.myStage2; nstage ++)
		{
			index = nstage - vis.myStage1;

			connection_name = "connection" + orderStage[nstage];
			//get the placement information of the next stage
			if(nstage == vis.myStage2) //The last stage in this node
			{
				GANode tempGANode = (GANode)(treenode.getUserObject());

				//Add a "bandwidth" to the configuration file
				xmlWriter.addPara(nstage, "bandwidth"+orderStage[nstage], 
					   "String", Integer.toString(tempGANode.real_dist));

				//We need to recurviely call WriteToConfigFile
				//and pass the parent node to the function
				DefaultMutableTreeNode parent = (DefaultMutableTreeNode)treenode.getParent();
				myConnectionValue = WriteToConfigFile(xmlWriter, parent, nDepth);
				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"+orderStage[nstage+1];
				//Add a "bandwidth" to the configuration file
				xmlWriter.addPara(nstage, "bandwidth"+orderStage[nstage], 
					   "String", "-1");
			}

			//Get the classname and location
			vis.strConnections[index] = myConnectionValue;
			if(nstage > numStages)
			{
				String specialclass_name = (String)XMLConfigurator.getParameter("special_class|class_name");
				String specialclass_location = (String)XMLConfigurator.getParameter("special_class|location");
				xmlWriter.addClass(nstage, specialclass_name, specialclass_location);
			}

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