Context Navigation

DCWormsUsers.java @ 1207

Revision 1207, 10.6 KB checked in by wojtekp, 11 years ago (diff)

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

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

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

Download in other formats:

Original Format