Context Navigation

← Previous Revision
Latest Revision
Next Revision →
Normal
Revision Log

GSSimUsersNew.java @ 104

Revision 104, 9.8 KB checked in by wojtekp, 13 years ago (diff)

Property svn:mime-type set to `text/plain`

Rev	Line
[104]	1	package test.rewolucja;
	2
	3	import eduni.simjava.Sim_event;
	4	import eduni.simjava.Sim_system;
	5	import gridsim.GridSim;
	6	import gridsim.GridSimTags;
	7	import gridsim.Gridlet;
	8	import gridsim.IO_data;
	9	import gridsim.gssim.GssimConstants;
	10	import gridsim.gssim.network.ExtendedGridSim;
	11	import gridsim.net.InfoPacket;
	12	import grms.shared.constants.BrokerConstants;
	13	import gssim.schedframe.scheduling.utils.JobDescription;
	14	import gssim.schedframe.scheduling.utils.TaskDescription;
	15
	16	import java.util.ArrayList;
	17	import java.util.HashSet;
	18	import java.util.Iterator;
	19	import java.util.List;
	20	import java.util.Map;
	21	import java.util.Set;
	22	import java.util.TreeMap;
	23
	24	import org.apache.commons.logging.Log;
	25	import org.apache.commons.logging.LogFactory;
	26	import org.joda.time.DateTime;
	27
	28	import schedframe.scheduling.Job;
	29	import schedframe.scheduling.JobInterface;
	30	import schedframe.scheduling.Task;
	31	import schedframe.scheduling.TaskInterface;
	32	import simulator.GenericUser;
	33	import simulator.utils.XsltTransformations;
	34	import simulator.workload.WorkloadLoader;
	35
	36	public class GSSimUsersNew extends ExtendedGridSim implements GenericUser {
	37
	38	/*A job generator, which produces jobs and tasks. These jobs are then sent by this entity /
	39	protected WorkloadLoader workloadLoader;
	40
	41	/** The name of the entity, to which the created tasks will be sent */
	42	protected String destName;
	43
	44	/** Indicates, that all tasks that returned to this object are finished */
	45	protected boolean allTasksAreFinished;
	46
	47	/** Stores the list of jobs, that have been returned to this object */
	48	protected List<JobInterface<?>> returnedJobs;
	49	protected Set<String> sendJobsIds;
	50	protected Set<String> returnedJobsIds;
	51
	52	protected XsltTransformations xsltTransformer;
	53
	54	/**
	55	* Indicates, that an error has occurred - it is used for debug purposes
	56	*/
	57	protected boolean error;
	58
	59	private static Log log = LogFactory.getLog(GSSimUsersNew.class);
	60
	61	/**
	62	* Constructs the users object with the given parameters
	63	* @param name the name of the users entity (must be unique across all entities in the whole simulation)
	64	* @param destinationName the name of the entity, to which the created tasks will be sent
	65	* @param jobGenerator the job generator, which produces jobs and tasks, that will be sent by this class
	66	* @throws Exception if any occurs (see {@link GridSim#GridSim(String, double)})
	67	*/
	68	public GSSimUsersNew(String name, String destinationName, WorkloadLoader workload) throws Exception {
	69	super(name, GssimConstants.DEFAULT_BAUD_RATE);
	70	this.workloadLoader = workload;
	71	destName = destinationName;
	72	allTasksAreFinished = true;
	73	error = false;
	74
	75	sendJobsIds = new HashSet<String>();
	76	returnedJobsIds = new HashSet<String>();
	77
	78	this.xsltTransformer = new XsltTransformations();
	79	}
	80
	81	@Override
	82	public void body() {
	83	sendJobs();
	84	collectJobs();
	85
	86	//GridSim dependent code for shutting down the simulation
	87	shutdownGridStatisticsEntity();
	88	terminateIOEntities();
	89	shutdownUserEntity();
	90	}
	91
	92	/**
	93	* Collects the jobs sent to broker(s)
	94	*/
	95	protected void collectJobs() {
	96	final int FACTOR = Math.min(10, workloadLoader.getTaskCount()); //the refresh rate of the gauge: at most 10 times
	97	final int denominator = workloadLoader.getTaskCount() / FACTOR;
	98	allTasksAreFinished = true;
	99	returnedJobs = new ArrayList<JobInterface<?>>();
	100	int counter = 0;
	101	int oldRemeinder = 0;
	102	boolean hundredPercent = false;
	103
	104	Sim_event ev = new Sim_event();
	105	sim_get_next(ev);
	106	while (Sim_system.running()) {
	107	switch (ev.get_tag()) {
	108	case GridSimTags.END_OF_SIMULATION:
	109	//no action
	110	break;
	111
	112	case GridSimTags.INFOPKT_SUBMIT:
	113	processPingRequest(ev);
	114	break;
	115
	116	case GridSimTags.GRIDLET_RETURN:
	117	JobInterface<?> returnedJob = (JobInterface<?>) ev.get_data();
	118
	119	String jobId = null;
	120	try {
	121	jobId = returnedJob.getId();
	122	} catch (NoSuchFieldException e) {
	123	// TODO Auto-generated catch block
	124	e.printStackTrace();
	125	}
	126
	127	if (returnedJobs.contains(returnedJob)) {
	128	if(log.isErrorEnabled())
	129	log.error("Received the same job twice (job ID " + jobId + ")");
	130	error = true;
	131	break;
	132	}
	133
	134	returnedJobs.add(returnedJob);
	135	returnedJobsIds.add(jobId);
	136
	137	if(returnedJob.getStatus() == BrokerConstants.JOB_STATUS_FINISHED) {
	138	if(log.isDebugEnabled())
	139	log.debug("Received finished job " + jobId);
	140
	141	} else {
	142	if(returnedJob.getStatus() == BrokerConstants.JOB_STATUS_CANCELED) {
	143	if(log.isWarnEnabled()){
	144	String str = "Warning! An uncomplished job (Job ID: "+jobId+") has returned to users. Job was canceled.";
	145	log.warn(str);
	146	}
	147	}
	148
	149	allTasksAreFinished = false;
	150	}
	151
	152	counter += returnedJob.getTaskCount();
	153	int remainder = (counter / denominator);
	154	if (remainder != oldRemeinder) {
	155	int gauge = ((counter * 100) / (workloadLoader.getTaskCount()));
	156	if(log.isInfoEnabled())
	157	log.info(gauge + "% ");
	158
	159	oldRemeinder = remainder;
	160	if (gauge == 100)
	161	hundredPercent = true;
	162	}
	163
	164	break;
	165	}
	166
	167	//if all the Gridlets have been collected
	168	if (counter == workloadLoader.getTaskCount()) {
	169	break;
	170	}
	171
	172	sim_get_next(ev);
	173	}
	174
	175	//if all the Gridlets have been collected
	176	if (counter == workloadLoader.getTaskCount()) {
	177	if (! hundredPercent) {
	178	if(log.isInfoEnabled())
	179	log.info("100%");
	180	}
	181	if(log.isInfoEnabled())
	182	log.info(get_name() + ": Received all "+workloadLoader.getJobCount()+" jobs and " + counter + " tasks");
	183	} else {
	184	if(log.isErrorEnabled())
	185	log.error(get_name() + ": ERROR DID NOT RECEIVED all tasks - some tasks were not finished! (received "+counter+" of "+workloadLoader.getTaskCount()+")");
	186	}
	187
	188	Iterator <String> itr = sendJobsIds.iterator();
	189	String jobId;
	190	if(log.isInfoEnabled()){
	191	log.info("Missing tasks: ");
	192	while(itr.hasNext()){
	193	jobId = itr.next();
	194	if(!returnedJobsIds.contains(jobId)){
	195	log.info(jobId + ", ");
	196	}
	197	}
	198	}
	199
	200	}
	201
	202	/**
	203	* Sends jobs to broker entities
	204	*/
	205	protected void sendJobs() {
	206
	207	Map<Long, List<Job>> jobTimes = new TreeMap<Long, List<Job>>();
	208	int destID = GridSim.getEntityId(destName);
	209	destID=GridSim.getEntityId("BROKER@COMPUTING_GRID_0");
	210	List<JobDescription> jobs = workloadLoader.getJobs();
	211	//TaskRequirements taskReq = new TaskRequirementsImpl();
	212	//double values[] = null;
	213
	214	for (JobDescription job : jobs) {
	215	long l_submissionTime = Long.MAX_VALUE;
	216	//pick the lowest submission time
	217	for (TaskDescription task : job) {
	218	if (task.getSubmissionTime() < l_submissionTime)
	219	l_submissionTime = task.getSubmissionTime();
	220	}
	221
	222	//store the submission time expressed in seconds after the simulation start time
	223	long submissionTime = l_submissionTime;
	224
	225	Job newJob = prepareJob(job, submissionTime);
	226
	227	List<Job> list = jobTimes.get(submissionTime);
	228	if (list == null) {
	229	list = new ArrayList<Job>();
	230	jobTimes.put(submissionTime, list);
	231	}
	232	list.add(newJob);
	233	}
	234	for (Long submissionTime : jobTimes.keySet()) {
	235	List<Job> list = jobTimes.get(submissionTime);
	236	for(int i = 0; i < list.size(); i++){
	237	this.sendJobsIds.add(list.get(i).getId());
	238	send(destID, submissionTime, GridSimTags.GRIDLET_SUBMIT, list.get(i));
	239	}
	240
	241	//send(output, submissionTime, GridSimTags.GRIDLET_SUBMIT, new IO_data(list, GssConstants.DEFAULT_GRIDLET_SIZE, destID));
	242	//send(destID, submissionTime, GridSimTags.GRIDLET_SUBMIT, list);
	243
	244	}
	245	System.out.println("finished sending jobs");
	246	}
	247
	248	public List<JobDescription> getAllSentJobs() {
	249	return (List<JobDescription>) workloadLoader.getJobs();
	250	}
	251
	252	public List<TaskDescription> getAllSentTasks() {
	253	List<TaskDescription> result = new ArrayList<TaskDescription>();
	254	List<JobDescription> sentJobs = getAllSentJobs();
	255	for (JobDescription job : sentJobs) {
	256	result.addAll(job);
	257	}
	258	return result;
	259	}
	260
	261	public List<JobInterface<?>> getAllReceivedJobs() {
	262	return returnedJobs;
	263	}
	264
	265	public int getFinishedTasksCount() {
	266	int result = 0;
	267	for (JobInterface<?> job : returnedJobs) {
	268	for (TaskInterface<?> task: job.getTask()) {
	269	if(task.getStatus() == Gridlet.SUCCESS)
	270	result++;
	271	}
	272	}
	273	return result;
	274	}
	275
	276	public String getUserName() {
	277	return get_name();
	278	}
	279
	280	public boolean isError() {
	281	return error;
	282	}
	283
	284	/**
	285	* Performs action concerning a ping request to this entity
	286	* @param ev the event object
	287	*/
	288	protected void processPingRequest(Sim_event ev) {
	289	InfoPacket pkt = (InfoPacket) ev.get_data();
	290	pkt.setTag(GridSimTags.INFOPKT_RETURN);
	291	pkt.setDestID(pkt.getSrcID());
	292
	293	// sends back to the sender
	294	send(output, GridSimTags.SCHEDULE_NOW, GridSimTags.INFOPKT_RETURN,
	295	new IO_data(pkt, pkt.getSize(), pkt.getSrcID()));
	296	}
	297
	298	protected Job prepareJob(JobDescription jobDescription, long submissionTime){
	299
	300	Job newJob = new Job(jobDescription.getJobId());
	301	DateTime submitionTime = new DateTime();
	302	submitionTime = submitionTime.plusMillis((int)submissionTime*1000);
	303	try {
	304
	305	// transform job description to resource requirements
	306
	307	newJob.setSenderId(this.get_id());
	308
	309	for (TaskDescription taskDescription : jobDescription) {
	310
	311	String xmlResReq = this.xsltTransformer.taskToResourceRequirements(
	312	taskDescription.getDocument(),
	313	jobDescription.getJobId(),
	314	taskDescription.getUserDn(),
	315	submitionTime);
	316
	317	Task newTask = new Task(xmlResReq);
	318	newTask.setSenderId(this.get_id());
	319	newTask.setStatus((int)BrokerConstants.TASK_STATUS_UNSUBMITTED);
	320	newTask.setLength(taskDescription.getTaskLength());
	321	newTask.setWorkloadLogWaitTime(taskDescription.getWorkloadLogWaitTime());
	322	// newTask.setSubmissionTime(taskDescription.getSubmissionTime());
	323
	324	newJob.add(newTask);
	325	}
	326
	327	jobDescription.discardUnused();
	328
	329	} catch (Exception e){
	330	log.error(e.getMessage());
	331	e.printStackTrace();
	332	}
	333
	334	return newJob;
	335	}
	336	}

Note: See TracBrowser for help on using the repository browser.

Download in other formats:

Original Format