You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by sh...@apache.org on 2015/12/21 18:31:21 UTC
airavata git commit: Handle job cancel with queued jobs
Repository: airavata
Updated Branches:
refs/heads/master 47949d3eb -> 18a4c31b4
Handle job cancel with queued jobs
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/18a4c31b
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/18a4c31b
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/18a4c31b
Branch: refs/heads/master
Commit: 18a4c31b4904448fef63a7e0ea578ddeb9df217f
Parents: 47949d3
Author: Shameera Rathnayaka <sh...@gmail.com>
Authored: Mon Dec 21 12:31:14 2015 -0500
Committer: Shameera Rathnayaka <sh...@gmail.com>
Committed: Mon Dec 21 12:31:14 2015 -0500
----------------------------------------------------------------------
.../airavata/gfac/core/context/TaskContext.java | 9 ++++++++
.../airavata/gfac/impl/GFacEngineImpl.java | 13 +-----------
.../gfac/monitor/email/EmailBasedMonitor.java | 22 +++++++++++++++-----
3 files changed, 27 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/airavata/blob/18a4c31b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/context/TaskContext.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/context/TaskContext.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/context/TaskContext.java
index ae92ba1..b0b13b0 100644
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/context/TaskContext.java
+++ b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/context/TaskContext.java
@@ -39,6 +39,7 @@ public class TaskContext {
private InputDataObjectType processInput;
private OutputDataObjectType processOutput;
private Object subTaskModel = null;
+ private boolean isCancel = false;
public TaskModel getTaskModel() {
return taskModel;
@@ -117,4 +118,12 @@ public class TaskContext {
}
return subTaskModel;
}
+
+ public boolean isCancel() {
+ return isCancel;
+ }
+
+ public void setCancel(boolean cancel) {
+ isCancel = cancel;
+ }
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/18a4c31b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacEngineImpl.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacEngineImpl.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacEngineImpl.java
index 8eb77e1..9e6a522 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacEngineImpl.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/GFacEngineImpl.java
@@ -627,18 +627,7 @@ public class GFacEngineImpl implements GFacEngine {
private void executeCancel(TaskContext taskContext, JobSubmissionTask jSTask) throws GFacException {
try {
JobStatus oldJobStatus = jSTask.cancel(taskContext);
-
-/* if (oldJobStatus != null && oldJobStatus.getJobState() == JobState.QUEUED) {
- ProcessContext pc = taskContext.getParentProcessContext();
- JobStatus newJobStatus = new JobStatus(JobState.CANCELED);
- newJobStatus.setReason("Job cancelled");
- newJobStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
- pc.getJobModel().setJobStatus(newJobStatus);
- GFacUtils.saveJobStatus(pc, pc.getJobModel());
- JobMonitor monitorService = Factory.getMonitorService(pc.getMonitorMode());
- monitorService.stopMonitor(pc.getJobModel().getJobId(), true);
- }*/
-
+ // If Job was in Queued state when cancel command runs, then there won't be any email from this job.
ProcessContext pc = taskContext.getParentProcessContext();
JobMonitor monitorService = Factory.getMonitorService(pc.getMonitorMode());
monitorService.canceledJob(pc.getJobModel().getJobId());
http://git-wip-us.apache.org/repos/asf/airavata/blob/18a4c31b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/monitor/email/EmailBasedMonitor.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/monitor/email/EmailBasedMonitor.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/monitor/email/EmailBasedMonitor.java
index 69fae9a..b529fb5 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/monitor/email/EmailBasedMonitor.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/monitor/email/EmailBasedMonitor.java
@@ -130,7 +130,18 @@ public class EmailBasedMonitor implements JobMonitor, Runnable{
TaskContext taskContext = jobMonitorMap.remove(jobId);
if (taskContext != null && runOutflow) {
try {
- GFacThreadPoolExecutor.getCachedThreadPool().execute(new GFacWorker(taskContext.getParentProcessContext()));
+ ProcessContext pc = taskContext.getParentProcessContext();
+ if (taskContext.isCancel()) {
+ // Moved job status to cancel
+ JobModel jobModel = pc.getJobModel();
+ JobStatus newJobStatus = new JobStatus(JobState.CANCELED);
+ newJobStatus.setReason("Moving job status to cancel, as we didn't see any email from this job " +
+ "for a while after execute job cancel command. This may happen if job was in queued state " +
+ "when we run the cancel command");
+ jobModel.setJobStatus(newJobStatus);
+ GFacUtils.saveJobStatus(pc, jobModel);
+ }
+ GFacThreadPoolExecutor.getCachedThreadPool().execute(new GFacWorker(pc));
} catch (GFacException e) {
log.info("[EJM]: Error while running output tasks", e);
}
@@ -317,13 +328,13 @@ public class EmailBasedMonitor implements JobMonitor, Runnable{
}
private void process(JobStatusResult jobStatusResult, TaskContext taskContext){
+ canceledJobs.remove(jobStatusResult.getJobId());
JobState resultState = jobStatusResult.getState();
- // TODO : update job state on process context
+ // TODO : update job state on process context
boolean runOutflowTasks = false;
- JobStatus jobStatus = new JobStatus();
- JobModel jobModel = taskContext.getParentProcessContext().getJobModel();
+ JobStatus jobStatus = new JobStatus();
+ JobModel jobModel = taskContext.getParentProcessContext().getJobModel();
String jobDetails = "JobName : " + jobStatusResult.getJobName() + ", JobId : " + jobStatusResult.getJobId();
- canceledJobs.remove(jobStatusResult.getJobId());
// TODO - Handle all other valid JobStates
if (resultState == JobState.COMPLETE) {
jobMonitorMap.remove(jobStatusResult.getJobId());
@@ -423,6 +434,7 @@ public class EmailBasedMonitor implements JobMonitor, Runnable{
} else {
TaskContext taskContext = jobMonitorMap.get(cancelJobIdWithFlag.getKey());
if (taskContext != null) {
+ taskContext.setCancel(true);
stopMonitor(cancelJobIdWithFlag.getKey(), true);
}
cancelJobIter.remove();